当前位置：首页 > ops >正文

探索分布式存储与通信：去中心化共享及通訊（DSAC）

ops 2025/6/29 6:02:50

在当今数字化时代，分布式系统的重要性愈发凸显。它不仅能提升数据的存储安全性和可靠性，还能增强通信的效率和隐私性。于是我做了这个去中心化共享及通訊的程序，它构建了一个强大的分布式存储和通信网络，下面我们就来详细了解其实现原理和核心功能。

程序概述

实现了一个基于点对点（P2P）网络的分布式存储和通信系统。该系统允许节点加入网络、共享文件、存储文件块以及进行加密通信。它采用了分布式哈希表（DHT）的思想，通过节点间的协作来存储和检索文件。

核心类与功能

1. `Node` 类

Node 类代表网络中的一个节点，它包含了节点的基本信息和状态。以下是该类的主要功能：

地址簿管理：维护一个地址簿，记录网络中其他节点的信息，包括主机名、端口号和最后活跃时间。
文件管理：存储文件的元数据和文件块，支持文件的注册和广播。
加密密钥管理：使用 SHA - 256 算法为每个节点生成共享加密密钥，确保通信的安全性。

class Node:def __init__(self, host, port, domain):self.host = hostself.port = portself.domain = domainself.address_book = {}  # {domain: (host, port, last_seen)}self.files = {}  # {file_hash: {'name': str, 'size': int, 'chunks': list}}self.stored_chunks = {}  # {chunk_hash: data}self.network_initialized = Falseself.is_bootstrap = Falseself.peer_lock = threading.Lock()self.file_lock = threading.Lock()self.shared_secrets = {}  # 存储加密密钥# 初始化时将自己加入地址簿self.add_peer(domain, host, port)

2. `NetworkManager` 类

NetworkManager 类负责节点的网络通信和管理，它处理节点的加入、消息传递、文件上传和下载等任务。以下是该类的主要功能：

服务器启动：启动一个 TCP 服务器，监听其他节点的连接请求。
节点加入网络：允许节点通过引导节点加入网络，并同步地址簿信息。
消息传递：支持直接消息和中继消息的发送和接收，确保通信的可靠性。
文件操作：实现文件的上传、下载和存储，将文件分割成块并存储在多个节点上。

class NetworkManager:def __init__(self, node, log_callback):self.node = nodeself.log_callback = log_callbackself.server_socket = Noneself.running = False# 启动节点状态检查self.status_check_timer = threading.Timer(PING_INTERVAL, self.check_peer_status)self.status_check_timer.daemon = Trueself.status_check_timer.start()def start_server(self):self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server_socket.bind((self.node.host, self.node.port))self.server_socket.listen(5)self.running = Trueserver_thread = threading.Thread(target=self._accept_connections)server_thread.daemon = Trueserver_thread.start()self.log_callback(f"服务器已启动在 {self.node.host}:{self.node.port}")

核心功能实现

1. 节点加入网络

节点通过向引导节点发送 join 请求来加入网络。引导节点收到请求后，会检查域名是否已存在，如果不存在则将新节点添加到地址簿，并广播新节点信息给其他节点。

def _handle_join(self, message, client_socket):domain = message['domain']host = message['host']port = message['port']# 检查域名是否已存在if domain in self.node.address_book:response = {'type': 'join_error','message': '域名已存在'}client_socket.send(json.dumps(response).encode())self.log_callback(f"拒绝加入: 域名 {domain} 已存在")return# 添加新节点到地址簿self.node.add_peer(domain, host, port)# 发送完整的地址簿给新节点response = {'type': 'join_ack','bootstrap': self.node.is_bootstrap,'address_book': self.node.get_all_peers_with_status()}client_socket.send(json.dumps(response).encode())# 通知日志self.log_callback(f"新节点加入: {domain} ({host}:{port})")# 广播新节点信息给所有已知节点（除了新节点自己）self._broadcast_new_node(domain, host, port)

2. 文件上传与下载

文件上传时，系统会将文件分割成多个块，并将每个块存储在多个节点上。同时，文件的元数据会被广播到网络中的其他节点。文件下载时，系统会首先获取文件的元数据，然后从各个节点收集文件块，最后验证文件的完整性并保存。

def upload_file(self, file_path, num_chunks=4, num_replicas=2):try:file_name = os.path.basename(file_path)file_size = os.path.getsize(file_path)# 读取文件内容with open(file_path, 'rb') as f:file_data = f.read()# 创建文件哈希file_hash = hashlib.sha256(file_data).hexdigest()# 分割文件为多个块chunks = []chunk_size = max(FILE_CHUNK_SIZE, (len(file_data) + num_chunks - 1) // num_chunks)for i in range(0, len(file_data), chunk_size):chunk = file_data[i:i+chunk_size]chunk_hash = hashlib.sha256(chunk).hexdigest()chunks.append(chunk_hash)# 存储块到多个节点self._store_chunk(chunk, chunk_hash, num_replicas)# 注册文件元数据metadata = {'name': file_name,'size': file_size,'chunks': chunks}self.node.register_file(file_hash, file_name, file_size, chunks)# 广播文件元数据self.node.broadcast_file_metadata(file_hash, metadata)# 通知日志self.log_callback(f"文件上传成功: {file_name} (哈希: {file_hash})")return file_hashexcept Exception as e:self.log_callback(f"文件上传失败: {str(e)}")return None

3. 加密通信

系统使用 AES 算法对消息进行加密，确保通信内容的隐私性。每个节点会根据双方的域名生成一个共享密钥，用于加密和解密消息。

def _generate_shared_key(self, target_domain):"""生成共享密钥 - 基于双方域名生成确定性密钥"""# 按字母顺序排序域名以确保双方生成相同密钥domains = sorted([self.node.domain, target_domain])key_str = f"{domains[0]}-{domains[1]}"# 使用SHA-256生成256位哈希，取前16字节作为AES-128密钥key = hashlib.sha256(key_str.encode()).digest()[:16]self.node.shared_secrets[target_domain] = keyreturn keydef _encrypt_message(self, message, key):iv = get_random_bytes(16)cipher = AES.new(key, AES.MODE_CFB, iv)encrypted = cipher.encrypt(message.encode())encrypted_base64 = base64.b64encode(encrypted).decode()while len(encrypted_base64) % 4 != 0:encrypted_base64 += '='iv_base64 = base64.b64encode(iv).decode()while len(iv_base64) % 4 != 0:iv_base64 += '='return encrypted_base64, iv_base64