分布式文件存储系统简单实现

分布式文件存储系统简单实现

分布式文件存储系统通过将数据分散存储在多个节点上,解决单机存储的容量、性能和可靠性瓶颈，其核心目标是实现数据的高可用性、可扩展性和负载均衡，本文以最小化实现为例，介绍基于Python的简单分布式文件存储系统设计与实现。

架构图示：

Client <--> Master <--> [Chunk Server 1, Chunk Server 2, ...]

核心功能实现

元数据管理

元数据服务器维护文件到分块的映射关系,采用键值对存储：

class MetaDataServer:
def __init__(self):
self.file_map = {} # {filename: [(chunk_id, node_address)]}
def register_file(self, filename, chunks):
self.file_map[filename] = chunks
def get_file_info(self, filename):
return self.file_map.get(filename, None)

数据分片与复制

• 分片策略：按固定大小（如4MB）分割文件，生成唯一分片ID。
• 副本机制：每个分片存储3个副本，采用哈希取模分配节点。

分片示例：

import hashlib
def get_node_for_chunk(chunk_id, nodes):
    hash_val = int(hashlib.md5(chunk_id.encode()).hexdigest(), 16)
    return nodes[hash_val % len(nodes)]

存储节点实现

存储节点提供HTTP接口管理分块数据,支持CRUD操作：

from flask import Flask, request, send_file
app = Flask(__name__)
storage_path = "/data/"
@app.route('/chunk/<chunk_id>', methods=['POST'])
def upload_chunk(chunk_id):
    with open(storage_path + chunk_id, 'wb') as f:
        f.write(request.data)
    return "OK"
@app.route('/chunk/<chunk_id>', methods=['GET'])
def download_chunk(chunk_id):
    return send_file(storage_path + chunk_id)

客户端交互流程

文件上传流程：

1. 客户端将文件分割为多个chunk
2. 向元数据服务器注册文件信息
3. 并行上传所有chunk至分配节点
4. 更新元数据服务器中的副本信息

文件下载流程：

1. 查询元数据服务器获取chunk位置
2. 并行下载所有chunk
3. 合并chunk还原完整文件

关键代码示例

元数据服务器API

from flask import Flask, request, jsonify
app = Flask(__name__)
metadata = MetaDataServer()
nodes = ["http://node1:5000", "http://node2:5000"]
@app.route('/register', methods=['POST'])
def register_file():
    data = request.json
    filename = data["filename"]
    chunks = [(f"chunk_{i}", get_node_for_chunk(f"chunk_{i}", nodes)) for i in range(len(data["chunks"]))]
    metadata.register_file(filename, chunks)
    return jsonify({"status": "success"})

客户端上传逻辑

import requests
def upload_file(file_path):
    # 分割文件
    chunks = split_file(file_path)
    # 注册元数据
    response = requests.post("http://master:8000/register", json={
        "filename": file_path,
        "chunks": [c.id for c in chunks]
    })
    # 上传分块
    for chunk in chunks:
        with open(chunk.path, 'rb') as f:
            requests.post(chunk.node + f"/chunk/{chunk.id}", data=f)
    print("Upload completed")

容错与扩展设计

节点故障处理

• 心跳检测：元数据服务器定期检查存储节点状态
• 自动副本重建：当检测到节点故障时，触发副本重分配

水平扩展

• 添加新节点：元数据服务器动态更新节点列表
• 数据再平衡：按一致性哈希重新分配部分分块

FAQs

Q1：如何保证数据一致性？
A1：采用强一致性模型，所有写入操作需等待元数据服务器确认，可通过分布式锁或版本号机制避免冲突。

Q2：系统最大存储容量如何计算？
A2：总容量 = 单节点存储空间 × 节点数量 × 副本数，例如10个节点、每个1TB、副本数3，则总容量为30TB（实际有效容量10TB）。