元数据管理: 单点、集群化与分离式架构

老马啸西风2025/9/7大约 19 分钟

元数据管理是分布式文件存储系统的核心组件之一，负责维护文件系统的命名空间、文件属性、权限信息以及文件与数据块的映射关系。在分布式环境中，元数据管理面临着高可用性、一致性、性能和扩展性等多重挑战。本章将深入探讨元数据管理的三种主要架构模式：单点架构、集群化架构和分离式架构，分析各自的优缺点和适用场景，并提供实际的实现方案和最佳实践。

2.3 元数据管理架构模式

2.3.1 单点架构（Single Point Architecture）

单点架构是最简单的元数据管理方式，采用单一的元数据服务器来管理所有元数据信息。这种架构在早期的分布式文件系统中较为常见，如早期版本的NFS和某些私有云存储系统。

2.3.1.1 架构特点

集中式管理：
- 所有元数据集中存储在单一服务器上
- 简化的数据一致性和管理逻辑
- 易于实现和维护
简单性：
- 架构简单，易于理解和实现
- 开发和测试成本较低
- 故障诊断相对容易
性能优势：
- 无分布式协调开销
- 元数据访问延迟较低
- 事务处理相对简单

2.3.1.2 实现示例

# 单点元数据管理器实现
import threading
import time
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum

class FileType(Enum):
    FILE = "file"
    DIRECTORY = "directory"
    SYMLINK = "symlink"

@dataclass
class Inode:
    inode_id: str
    file_type: FileType
    size: int
    permissions: int
    uid: int
    gid: int
    atime: float
    mtime: float
    ctime: float
    blocks: List[str]
    extended_attributes: Dict[str, str]
    symlink_target: Optional[str] = None

class SinglePointMetadataManager:
    def __init__(self):
        self.inodes: Dict[str, Inode] = {}
        self.namespace: Dict[str, Dict[str, str]] = {}  # parent_path -> {name -> inode_id}
        self.lock = threading.RWLock()  # 读写锁
        self.next_inode_id = 1
        self.root_inode = self._create_inode(FileType.DIRECTORY)
        self.namespace["/"] = {}
    
    def _create_inode(self, file_type: FileType) -> Inode:
        """创建新的inode"""
        inode = Inode(
            inode_id=f"inode_{self.next_inode_id}",
            file_type=file_type,
            size=0,
            permissions=0o644,
            uid=0,
            gid=0,
            atime=time.time(),
            mtime=time.time(),
            ctime=time.time(),
            blocks=[],
            extended_attributes={}
        )
        self.next_inode_id += 1
        return inode
    
    def create_file(self, path: str, file_type: FileType = FileType.FILE) -> bool:
        """创建文件或目录"""
        with self.lock.write_lock():
            try:
                # 解析路径
                parts = path.strip("/").split("/")
                filename = parts[-1]
                parent_path = "/" + "/".join(parts[:-1]) if len(parts) > 1 else "/"
                
                # 检查父目录是否存在
                if parent_path not in self.namespace:
                    return False
                
                # 检查文件是否已存在
                if filename in self.namespace.get(parent_path, {}):
                    return False
                
                # 创建inode
                inode = self._create_inode(file_type)
                self.inodes[inode.inode_id] = inode
                
                # 更新命名空间
                if parent_path not in self.namespace:
                    self.namespace[parent_path] = {}
                self.namespace[parent_path][filename] = inode.inode_id
                
                # 如果是目录，初始化其命名空间
                if file_type == FileType.DIRECTORY:
                    self.namespace[path] = {}
                
                return True
            except Exception as e:
                print(f"创建文件失败: {e}")
                return False
    
    def lookup_inode(self, path: str) -> Optional[Inode]:
        """查找文件inode"""
        with self.lock.read_lock():
            try:
                # 解析路径
                if path == "/":
                    return self.root_inode
                
                parts = path.strip("/").split("/")
                current_inode_id = "inode_0"  # 根inode ID
                
                for part in parts:
                    if not part:
                        continue
                    
                    # 查找当前目录下的文件
                    dir_entries = self.namespace.get(self._get_path_from_inode(current_inode_id), {})
                    if part not in dir_entries:
                        return None
                    
                    current_inode_id = dir_entries[part]
                
                return self.inodes.get(current_inode_id)
            except Exception as e:
                print(f"查找inode失败: {e}")
                return None
    
    def _get_path_from_inode(self, inode_id: str) -> str:
        """根据inode ID获取路径（简化实现）"""
        # 在实际实现中，需要维护inode到路径的映射
        # 这里简化处理
        return "/"
    
    def delete_file(self, path: str) -> bool:
        """删除文件"""
        with self.lock.write_lock():
            try:
                # 解析路径
                parts = path.strip("/").split("/")
                filename = parts[-1]
                parent_path = "/" + "/".join(parts[:-1]) if len(parts) > 1 else "/"
                
                # 检查父目录是否存在
                if parent_path not in self.namespace:
                    return False
                
                # 检查文件是否存在
                parent_entries = self.namespace[parent_path]
                if filename not in parent_entries:
                    return False
                
                # 获取inode ID
                inode_id = parent_entries[filename]
                inode = self.inodes.get(inode_id)
                
                # 如果是目录，检查是否为空
                if inode and inode.file_type == FileType.DIRECTORY:
                    if self.namespace.get(path, {}):
                        print("目录非空，无法删除")
                        return False
                
                # 删除inode和命名空间条目
                if inode_id in self.inodes:
                    del self.inodes[inode_id]
                
                if filename in parent_entries:
                    del parent_entries[filename]
                
                # 如果是目录，删除其命名空间
                if path in self.namespace:
                    del self.namespace[path]
                
                return True
            except Exception as e:
                print(f"删除文件失败: {e}")
                return False
    
    def get_metadata_stats(self) -> Dict:
        """获取元数据统计信息"""
        with self.lock.read_lock():
            return {
                "total_inodes": len(self.inodes),
                "total_directories": len([inode for inode in self.inodes.values() if inode.file_type == FileType.DIRECTORY]),
                "total_files": len([inode for inode in self.inodes.values() if inode.file_type == FileType.FILE]),
                "namespace_entries": sum(len(entries) for entries in self.namespace.values())
            }

# 读写锁实现
class RWLock:
    def __init__(self):
        self._read_ready = threading.Condition(threading.RLock())
        self._readers = 0
    
    def read_lock(self):
        return ReadLock(self)
    
    def write_lock(self):
        return WriteLock(self)

class ReadLock:
    def __init__(self, rwlock):
        self.rwlock = rwlock
    
    def __enter__(self):
        self.rwlock._read_ready.acquire()
        try:
            self.rwlock._readers += 1
        finally:
            self.rwlock._read_ready.release()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.rwlock._read_ready.acquire()
        try:
            self.rwlock._readers -= 1
            if self.rwlock._readers == 0:
                self.rwlock._read_ready.notifyAll()
        finally:
            self.rwlock._read_ready.release()

class WriteLock:
    def __init__(self, rwlock):
        self.rwlock = rwlock
    
    def __enter__(self):
        self.rwlock._read_ready.acquire()
        while self.rwlock._readers > 0:
            self.rwlock._read_ready.wait()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.rwlock._read_ready.release()

# 使用示例
def test_single_point_metadata_manager():
    manager = SinglePointMetadataManager()
    
    # 创建目录结构
    print("创建目录结构...")
    manager.create_file("/home", FileType.DIRECTORY)
    manager.create_file("/home/user1", FileType.DIRECTORY)
    manager.create_file("/home/user1/documents", FileType.DIRECTORY)
    manager.create_file("/home/user1/documents/file1.txt")
    manager.create_file("/home/user1/documents/file2.txt")
    manager.create_file("/var", FileType.DIRECTORY)
    manager.create_file("/var/log", FileType.DIRECTORY)
    manager.create_file("/var/log/system.log")
    
    # 查看统计信息
    stats = manager.get_metadata_stats()
    print(f"元数据统计: {stats}")
    
    # 查找文件
    print("\n查找文件...")
    inode = manager.lookup_inode("/home/user1/documents/file1.txt")
    if inode:
        print(f"找到文件: {inode.inode_id}, 类型: {inode.file_type.value}, 大小: {inode.size}")
    else:
        print("文件未找到")
    
    # 删除文件
    print("\n删除文件...")
    result = manager.delete_file("/var/log/system.log")
    print(f"删除结果: {'成功' if result else '失败'}")
    
    # 再次查看统计信息
    stats = manager.get_metadata_stats()
    print(f"删除后元数据统计: {stats}")

# 运行测试
if __name__ == "__main__":
    test_single_point_metadata_manager()

2.3.1.3 优缺点分析

优点：

实现简单：架构简单，易于开发和维护
一致性保证：单点管理使得数据一致性容易保证
性能较好：无分布式协调开销，访问延迟较低
成本较低：只需要维护单一服务器

缺点：

单点故障：元数据服务器故障会导致整个系统不可用
扩展性限制：难以水平扩展，性能瓶颈明显
容量限制：单台服务器的存储和处理能力有限
维护困难：升级和维护期间系统不可用

2.3.1.4 适用场景

单点架构适用于以下场景：

小规模部署环境
对一致性要求极高的场景
开发测试环境
对成本敏感且规模较小的应用

2.3.2 集群化架构（Clustered Architecture）

集群化架构通过多个元数据服务器协同工作来管理元数据，提高了系统的可用性和扩展性。这种架构在现代分布式文件系统中广泛应用，如HDFS、Ceph等。

2.3.2.1 架构特点

高可用性：
- 通过主备或多主模式实现高可用
- 支持故障自动检测和恢复
- 消除单点故障风险
可扩展性：
- 支持水平扩展
- 可以动态添加或移除节点
- 负载可以分布到多个节点
复杂性：
- 需要实现分布式一致性协议
- 节点间协调机制复杂
- 故障处理逻辑复杂

2.3.2.2 实现示例

// 集群化元数据管理器实现
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.*;
import java.util.concurrent.atomic.*;

// 节点状态枚举
enum NodeStatus {
    ACTIVE, STANDBY, FAULTY, MAINTENANCE
}

// 元数据节点类
class MetadataNode {
    private String nodeId;
    private String host;
    private int port;
    private NodeStatus status;
    private long lastHeartbeat;
    private Map<String, Inode> inodes;
    private Map<String, Map<String, String>> namespace;
    private ReadWriteLock lock;
    
    public MetadataNode(String nodeId, String host, int port) {
        this.nodeId = nodeId;
        this.host = host;
        this.port = port;
        this.status = NodeStatus.STANDBY;
        this.lastHeartbeat = System.currentTimeMillis();
        this.inodes = new ConcurrentHashMap<>();
        this.namespace = new ConcurrentHashMap<>();
        this.lock = new ReentrantReadWriteLock();
    }
    
    // Getters and Setters
    public String getNodeId() { return nodeId; }
    public String getHost() { return host; }
    public int getPort() { return port; }
    public NodeStatus getStatus() { return status; }
    public void setStatus(NodeStatus status) { this.status = status; }
    public long getLastHeartbeat() { return lastHeartbeat; }
    public void updateHeartbeat() { this.lastHeartbeat = System.currentTimeMillis(); }
    
    public Map<String, Inode> getInodes() { return inodes; }
    public Map<String, Map<String, String>> getNamespace() { return namespace; }
    public ReadWriteLock getLock() { return lock; }
}

// 集群化元数据管理器
public class ClusteredMetadataManager {
    private List<MetadataNode> nodes;
    private MetadataNode masterNode;
    private ScheduledExecutorService heartbeatChecker;
    private AtomicInteger nextInodeId;
    private static final long HEARTBEAT_TIMEOUT = 30000; // 30秒
    
    public ClusteredMetadataManager() {
        this.nodes = new ArrayList<>();
        this.nextInodeId = new AtomicInteger(1);
        this.heartbeatChecker = Executors.newScheduledThreadPool(1);
        
        // 启动心跳检查
        this.heartbeatChecker.scheduleAtFixedRate(this::checkHeartbeats, 0, 10, TimeUnit.SECONDS);
    }
    
    public void addNode(String nodeId, String host, int port) {
        MetadataNode node = new MetadataNode(nodeId, host, port);
        nodes.add(node);
        
        // 如果还没有主节点，选举第一个节点为主节点
        if (masterNode == null) {
            masterNode = node;
            node.setStatus(NodeStatus.ACTIVE);
        }
    }
    
    public boolean createFile(String path, FileType fileType) {
        if (masterNode == null) {
            System.err.println("没有可用的主节点");
            return false;
        }
        
        // 在主节点上执行创建操作
        return createFileOnNode(masterNode, path, fileType);
    }
    
    private boolean createFileOnNode(MetadataNode node, String path, FileType fileType) {
        Lock writeLock = node.getLock().writeLock();
        writeLock.lock();
        try {
            // 解析路径
            String[] parts = path.replaceAll("^/+", "").split("/");
            String filename = parts[parts.length - 1];
            String parentPath = parts.length > 1 ? 
                "/" + String.join("/", Arrays.copyOf(parts, parts.length - 1)) : "/";
            
            // 检查父目录是否存在
            if (!node.getNamespace().containsKey(parentPath)) {
                // 创建父目录
                createParentDirectories(node, parentPath);
            }
            
            // 检查文件是否已存在
            Map<String, String> parentEntries = node.getNamespace().get(parentPath);
            if (parentEntries != null && parentEntries.containsKey(filename)) {
                return false;
            }
            
            // 创建inode
            String inodeId = "inode_" + nextInodeId.getAndIncrement();
            Inode inode = new Inode(
                inodeId, fileType, 0, 0644, 0, 0,
                System.currentTimeMillis(), System.currentTimeMillis(), System.currentTimeMillis(),
                new ArrayList<>(), new HashMap<>()
            );
            
            node.getInodes().put(inodeId, inode);
            
            // 更新命名空间
            if (!node.getNamespace().containsKey(parentPath)) {
                node.getNamespace().put(parentPath, new HashMap<>());
            }
            node.getNamespace().get(parentPath).put(filename, inodeId);
            
            // 如果是目录，初始化其命名空间
            if (fileType == FileType.DIRECTORY) {
                node.getNamespace().put(path, new HashMap<>());
            }
            
            // 同步到备节点（简化实现）
            syncToStandbyNodes(node, "CREATE", path, inode);
            
            return true;
        } catch (Exception e) {
            System.err.println("创建文件失败: " + e.getMessage());
            return false;
        } finally {
            writeLock.unlock();
        }
    }
    
    private void createParentDirectories(MetadataNode node, String parentPath) {
        String[] parts = parentPath.replaceAll("^/+", "").split("/");
        StringBuilder currentPath = new StringBuilder("/");
        
        for (String part : parts) {
            if (part.isEmpty()) continue;
            
            if (currentPath.length() > 1) {
                currentPath.append("/");
            }
            currentPath.append(part);
            
            String path = currentPath.toString();
            if (!node.getNamespace().containsKey(path)) {
                String inodeId = "inode_" + nextInodeId.getAndIncrement();
                Inode inode = new Inode(
                    inodeId, FileType.DIRECTORY, 0, 0755, 0, 0,
                    System.currentTimeMillis(), System.currentTimeMillis(), System.currentTimeMillis(),
                    new ArrayList<>(), new HashMap<>()
                );
                node.getInodes().put(inodeId, inode);
                node.getNamespace().put(path, new HashMap<>());
                
                // 更新父目录的条目
                if (path.lastIndexOf("/") > 0) {
                    String parentDir = path.substring(0, path.lastIndexOf("/"));
                    if (parentDir.isEmpty()) parentDir = "/";
                    String dirName = path.substring(path.lastIndexOf("/") + 1);
                    
                    if (!node.getNamespace().containsKey(parentDir)) {
                        node.getNamespace().put(parentDir, new HashMap<>());
                    }
                    node.getNamespace().get(parentDir).put(dirName, inodeId);
                }
            }
        }
    }
    
    public Inode lookupInode(String path) {
        if (masterNode == null) {
            System.err.println("没有可用的主节点");
            return null;
        }
        
        Lock readLock = masterNode.getLock().readLock();
        readLock.lock();
        try {
            if (path.equals("/")) {
                // 返回根目录inode
                for (Inode inode : masterNode.getInodes().values()) {
                    if (inode.getFileType() == FileType.DIRECTORY && 
                        masterNode.getNamespace().containsKey("/") &&
                        masterNode.getNamespace().get("/").isEmpty()) {
                        return inode;
                    }
                }
                return null;
            }
            
            String[] parts = path.replaceAll("^/+", "").split("/");
            String currentInodeId = findRootInodeId(masterNode);
            
            for (String part : parts) {
                if (part.isEmpty()) continue;
                
                Map<String, String> dirEntries = masterNode.getNamespace().get(getPathFromInode(masterNode, currentInodeId));
                if (dirEntries == null || !dirEntries.containsKey(part)) {
                    return null;
                }
                
                currentInodeId = dirEntries.get(part);
            }
            
            return masterNode.getInodes().get(currentInodeId);
        } finally {
            readLock.unlock();
        }
    }
    
    private String findRootInodeId(MetadataNode node) {
        // 简化实现，实际应该维护根inode的引用
        for (Map.Entry<String, Inode> entry : node.getInodes().entrySet()) {
            if (entry.getValue().getFileType() == FileType.DIRECTORY) {
                // 这里简化处理，实际需要更精确的判断
                return entry.getKey();
            }
        }
        return null;
    }
    
    private String getPathFromInode(MetadataNode node, String inodeId) {
        // 简化实现，实际应该维护inode到路径的映射
        return "/";
    }
    
    private void syncToStandbyNodes(MetadataNode master, String operation, String path, Inode inode) {
        // 同步到备节点（简化实现）
        for (MetadataNode node : nodes) {
            if (node != master && node.getStatus() == NodeStatus.STANDBY) {
                // 这里应该实现实际的同步逻辑
                System.out.println("同步操作 " + operation + " 到节点 " + node.getNodeId());
            }
        }
    }
    
    private void checkHeartbeats() {
        long currentTime = System.currentTimeMillis();
        boolean masterFailed = false;
        
        for (MetadataNode node : nodes) {
            if (currentTime - node.getLastHeartbeat() > HEARTBEAT_TIMEOUT) {
                if (node.getStatus() == NodeStatus.ACTIVE) {
                    System.out.println("主节点 " + node.getNodeId() + " 心跳超时");
                    masterFailed = true;
                    node.setStatus(NodeStatus.FAULTY);
                } else if (node.getStatus() == NodeStatus.STANDBY) {
                    System.out.println("备节点 " + node.getNodeId() + " 心跳超时");
                    node.setStatus(NodeStatus.FAULTY);
                }
            }
        }
        
        // 如果主节点故障，进行故障转移
        if (masterFailed) {
            failover();
        }
    }
    
    private void failover() {
        // 选择新的主节点
        for (MetadataNode node : nodes) {
            if (node.getStatus() == NodeStatus.STANDBY) {
                masterNode = node;
                node.setStatus(NodeStatus.ACTIVE);
                System.out.println("故障转移完成，新主节点: " + node.getNodeId());
                break;
            }
        }
    }
    
    public void updateHeartbeat(String nodeId) {
        for (MetadataNode node : nodes) {
            if (node.getNodeId().equals(nodeId)) {
                node.updateHeartbeat();
                break;
            }
        }
    }
    
    public Map<String, Object> getClusterStats() {
        Map<String, Object> stats = new HashMap<>();
        stats.put("totalNodes", nodes.size());
        stats.put("activeNodes", nodes.stream().filter(n -> n.getStatus() == NodeStatus.ACTIVE).count());
        stats.put("standbyNodes", nodes.stream().filter(n -> n.getStatus() == NodeStatus.STANDBY).count());
        stats.put("faultyNodes", nodes.stream().filter(n -> n.getStatus() == NodeStatus.FAULTY).count());
        
        if (masterNode != null) {
            Lock readLock = masterNode.getLock().readLock();
            readLock.lock();
            try {
                stats.put("totalInodes", masterNode.getInodes().size());
                stats.put("totalDirectories", masterNode.getInodes().values().stream()
                    .filter(inode -> inode.getFileType() == FileType.DIRECTORY).count());
                stats.put("totalFiles", masterNode.getInodes().values().stream()
                    .filter(inode -> inode.getFileType() == FileType.FILE).count());
            } finally {
                readLock.unlock();
            }
        }
        
        return stats;
    }
    
    // 关闭资源
    public void shutdown() {
        if (heartbeatChecker != null) {
            heartbeatChecker.shutdown();
        }
    }
}

// 文件类型枚举
enum FileType {
    FILE, DIRECTORY, SYMLINK
}

// Inode类
class Inode {
    private String inodeId;
    private FileType fileType;
    private long size;
    private int permissions;
    private int uid;
    private int gid;
    private long atime;
    private long mtime;
    private long ctime;
    private List<String> blocks;
    private Map<String, String> extendedAttributes;
    private String symlinkTarget;
    
    public Inode(String inodeId, FileType fileType, long size, int permissions,
                 int uid, int gid, long atime, long mtime, long ctime,
                 List<String> blocks, Map<String, String> extendedAttributes) {
        this.inodeId = inodeId;
        this.fileType = fileType;
        this.size = size;
        this.permissions = permissions;
        this.uid = uid;
        this.gid = gid;
        this.atime = atime;
        this.mtime = mtime;
        this.ctime = ctime;
        this.blocks = blocks;
        this.extendedAttributes = extendedAttributes;
    }
    
    // Getters and Setters
    public String getInodeId() { return inodeId; }
    public FileType getFileType() { return fileType; }
    public long getSize() { return size; }
    public int getPermissions() { return permissions; }
    public int getUid() { return uid; }
    public int getGid() { return gid; }
    public long getAtime() { return atime; }
    public long getMtime() { return mtime; }
    public long getCtime() { return ctime; }
    public List<String> getBlocks() { return blocks; }
    public Map<String, String> getExtendedAttributes() { return extendedAttributes; }
    public String getSymlinkTarget() { return symlinkTarget; }
    public void setSymlinkTarget(String symlinkTarget) { this.symlinkTarget = symlinkTarget; }
}

// 使用示例
class ClusteredMetadataManagerDemo {
    public static void main(String[] args) {
        ClusteredMetadataManager manager = new ClusteredMetadataManager();
        
        // 添加节点
        manager.addNode("node-001", "192.168.1.10", 8080);
        manager.addNode("node-002", "192.168.1.11", 8080);
        manager.addNode("node-003", "192.168.1.12", 8080);
        
        // 更新心跳
        manager.updateHeartbeat("node-001");
        manager.updateHeartbeat("node-002");
        manager.updateHeartbeat("node-003");
        
        // 创建目录结构
        System.out.println("创建目录结构...");
        manager.createFile("/home", FileType.DIRECTORY);
        manager.createFile("/home/user1", FileType.DIRECTORY);
        manager.createFile("/home/user1/documents", FileType.DIRECTORY);
        manager.createFile("/home/user1/documents/file1.txt", FileType.FILE);
        manager.createFile("/home/user1/documents/file2.txt", FileType.FILE);
        manager.createFile("/var", FileType.DIRECTORY);
        manager.createFile("/var/log", FileType.DIRECTORY);
        manager.createFile("/var/log/system.log", FileType.FILE);
        
        // 查看统计信息
        Map<String, Object> stats = manager.getClusterStats();
        System.out.println("集群统计信息: " + stats);
        
        // 查找文件
        System.out.println("\n查找文件...");
        Inode inode = manager.lookupInode("/home/user1/documents/file1.txt");
        if (inode != null) {
            System.out.println("找到文件: " + inode.getInodeId() + 
                             ", 类型: " + inode.getFileType() + 
                             ", 大小: " + inode.getSize());
        } else {
            System.out.println("文件未找到");
        }
        
        // 关闭资源
        manager.shutdown();
    }
}

2.3.2.3 优缺点分析

优点：

高可用性：通过主备或多主模式消除单点故障
良好扩展性：支持水平扩展，可以动态添加节点
负载分散：负载可以分布到多个节点上
容错能力：部分节点故障不影响整体服务

缺点：

实现复杂：需要实现分布式一致性协议
一致性开销：节点间同步带来额外开销
网络依赖：强依赖网络通信
运维复杂：集群管理相对复杂

2.3.2.4 适用场景

集群化架构适用于以下场景：

中大规模部署环境
对高可用性要求较高的场景
需要水平扩展的系统
生产环境的关键业务系统

2.3.3 分离式架构（Separation Architecture）

分离式架构将元数据管理与数据存储完全分离，元数据服务专注于元数据的管理，而数据服务专注于数据的存储和访问。这种架构在现代云存储系统中较为常见，如Amazon S3、Google Cloud Storage等。

2.3.3.1 架构特点

职责分离：
- 元数据服务和数据服务职责明确
- 各自独立扩展和优化
- 降低系统复杂性
独立扩展：
- 元数据服务和数据服务可以独立扩展
- 根据各自负载特点进行优化
- 提高资源利用率
技术专业化：
- 元数据服务可以采用专门的存储技术
- 数据服务可以针对大块数据进行优化
- 提高各自的专业化水平

2.3.3.2 实现示例

# 分离式元数据管理器实现
import asyncio
import aiohttp
import json
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import time

class ObjectType(Enum):
    BUCKET = "bucket"
    OBJECT = "object"

@dataclass
class ObjectMetadata:
    key: str
    bucket: str
    size: int
    etag: str
    content_type: str
    created_time: float
    modified_time: float
    owner: str
    permissions: str
    storage_class: str
    extended_attributes: Dict[str, str]

class SeparatedMetadataService:
    def __init__(self, metadata_store_url: str):
        self.metadata_store_url = metadata_store_url
        self.session = None
        self.cache = {}
        self.cache_ttl = 300  # 5分钟缓存
    
    async def init_session(self):
        """初始化HTTP会话"""
        if self.session is None:
            self.session = aiohttp.ClientSession()
    
    async def close_session(self):
        """关闭HTTP会话"""
        if self.session:
            await self.session.close()
            self.session = None
    
    async def create_bucket(self, bucket_name: str, owner: str) -> bool:
        """创建存储桶"""
        await self.init_session()
        
        metadata = ObjectMetadata(
            key="",
            bucket=bucket_name,
            size=0,
            etag="",
            content_type="",
            created_time=time.time(),
            modified_time=time.time(),
            owner=owner,
            permissions="private",
            storage_class="STANDARD",
            extended_attributes={}
        )
        
        try:
            async with self.session.post(
                f"{self.metadata_store_url}/buckets",
                json=asdict(metadata)
            ) as response:
                if response.status == 201:
                    # 清除缓存
                    cache_key = f"bucket:{bucket_name}"
                    if cache_key in self.cache:
                        del self.cache[cache_key]
                    return True
                else:
                    print(f"创建存储桶失败: {response.status}")
                    return False
        except Exception as e:
            print(f"创建存储桶异常: {e}")
            return False
    
    async def put_object_metadata(self, bucket_name: str, object_key: str, 
                                 metadata: ObjectMetadata) -> bool:
        """存储对象元数据"""
        await self.init_session()
        
        try:
            async with self.session.put(
                f"{self.metadata_store_url}/buckets/{bucket_name}/objects/{object_key}",
                json=asdict(metadata)
            ) as response:
                if response.status in [200, 201]:
                    # 更新缓存
                    cache_key = f"object:{bucket_name}:{object_key}"
                    self.cache[cache_key] = {
                        "data": metadata,
                        "timestamp": time.time()
                    }
                    return True
                else:
                    print(f"存储对象元数据失败: {response.status}")
                    return False
        except Exception as e:
            print(f"存储对象元数据异常: {e}")
            return False
    
    async def get_object_metadata(self, bucket_name: str, object_key: str) -> Optional[ObjectMetadata]:
        """获取对象元数据"""
        await self.init_session()
        
        # 检查缓存
        cache_key = f"object:{bucket_name}:{object_key}"
        if cache_key in self.cache:
            cached = self.cache[cache_key]
            if time.time() - cached["timestamp"] < self.cache_ttl:
                return cached["data"]
            else:
                # 缓存过期，删除
                del self.cache[cache_key]
        
        try:
            async with self.session.get(
                f"{self.metadata_store_url}/buckets/{bucket_name}/objects/{object_key}"
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    metadata = ObjectMetadata(**data)
                    
                    # 更新缓存
                    self.cache[cache_key] = {
                        "data": metadata,
                        "timestamp": time.time()
                    }
                    
                    return metadata
                elif response.status == 404:
                    return None
                else:
                    print(f"获取对象元数据失败: {response.status}")
                    return None
        except Exception as e:
            print(f"获取对象元数据异常: {e}")
            return None
    
    async def delete_object_metadata(self, bucket_name: str, object_key: str) -> bool:
        """删除对象元数据"""
        await self.init_session()
        
        try:
            async with self.session.delete(
                f"{self.metadata_store_url}/buckets/{bucket_name}/objects/{object_key}"
            ) as response:
                if response.status in [200, 204]:
                    # 清除缓存
                    cache_key = f"object:{bucket_name}:{object_key}"
                    if cache_key in self.cache:
                        del self.cache[cache_key]
                    return True
                else:
                    print(f"删除对象元数据失败: {response.status}")
                    return False
        except Exception as e:
            print(f"删除对象元数据异常: {e}")
            return False
    
    async def list_objects(self, bucket_name: str, prefix: str = "", 
                          max_keys: int = 1000) -> List[ObjectMetadata]:
        """列出对象"""
        await self.init_session()
        
        params = {
            "prefix": prefix,
            "max-keys": max_keys
        }
        
        try:
            async with self.session.get(
                f"{self.metadata_store_url}/buckets/{bucket_name}/objects",
                params=params
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    objects = [ObjectMetadata(**obj) for obj in data.get("objects", [])]
                    return objects
                else:
                    print(f"列出对象失败: {response.status}")
                    return []
        except Exception as e:
            print(f"列出对象异常: {e}")
            return []
    
    async def get_bucket_metadata(self, bucket_name: str) -> Optional[ObjectMetadata]:
        """获取存储桶元数据"""
        await self.init_session()
        
        # 检查缓存
        cache_key = f"bucket:{bucket_name}"
        if cache_key in self.cache:
            cached = self.cache[cache_key]
            if time.time() - cached["timestamp"] < self.cache_ttl:
                return cached["data"]
            else:
                # 缓存过期，删除
                del self.cache[cache_key]
        
        try:
            async with self.session.get(
                f"{self.metadata_store_url}/buckets/{bucket_name}"
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    metadata = ObjectMetadata(**data)
                    
                    # 更新缓存
                    self.cache[cache_key] = {
                        "data": metadata,
                        "timestamp": time.time()
                    }
                    
                    return metadata
                elif response.status == 404:
                    return None
                else:
                    print(f"获取存储桶元数据失败: {response.status}")
                    return None
        except Exception as e:
            print(f"获取存储桶元数据异常: {e}")
            return None

# 数据服务接口
class DataServiceInterface:
    def __init__(self, data_service_url: str):
        self.data_service_url = data_service_url
        self.session = None
    
    async def init_session(self):
        """初始化HTTP会话"""
        if self.session is None:
            self.session = aiohttp.ClientSession()
    
    async def close_session(self):
        """关闭HTTP会话"""
        if self.session:
            await self.session.close()
            self.session = None
    
    async def upload_object(self, bucket_name: str, object_key: str, 
                           data: bytes, metadata: ObjectMetadata) -> bool:
        """上传对象数据"""
        await self.init_session()
        
        try:
            # 这里简化处理，实际应该分块上传大文件
            async with self.session.put(
                f"{self.data_service_url}/buckets/{bucket_name}/objects/{object_key}",
                data=data,
                headers={
                    "Content-Type": metadata.content_type,
                    "Content-Length": str(len(data))
                }
            ) as response:
                return response.status in [200, 201]
        except Exception as e:
            print(f"上传对象数据异常: {e}")
            return False
    
    async def download_object(self, bucket_name: str, object_key: str) -> Optional[bytes]:
        """下载对象数据"""
        await self.init_session()
        
        try:
            async with self.session.get(
                f"{self.data_service_url}/buckets/{bucket_name}/objects/{object_key}"
            ) as response:
                if response.status == 200:
                    data = await response.read()
                    return data
                elif response.status == 404:
                    return None
                else:
                    print(f"下载对象数据失败: {response.status}")
                    return None
        except Exception as e:
            print(f"下载对象数据异常: {e}")
            return None
    
    async def delete_object(self, bucket_name: str, object_key: str) -> bool:
        """删除对象数据"""
        await self.init_session()
        
        try:
            async with self.session.delete(
                f"{self.data_service_url}/buckets/{bucket_name}/objects/{object_key}"
            ) as response:
                return response.status in [200, 204]
        except Exception as e:
            print(f"删除对象数据异常: {e}")
            return False

# 分离式存储系统
class SeparatedStorageSystem:
    def __init__(self, metadata_service_url: str, data_service_url: str):
        self.metadata_service = SeparatedMetadataService(metadata_service_url)
        self.data_service = DataServiceInterface(data_service_url)
    
    async def put_object(self, bucket_name: str, object_key: str, 
                        data: bytes, content_type: str = "application/octet-stream",
                        owner: str = "anonymous") -> bool:
        """存储对象"""
        # 首先上传数据
        metadata = ObjectMetadata(
            key=object_key,
            bucket=bucket_name,
            size=len(data),
            etag="",  # 简化处理，实际应该计算MD5
            content_type=content_type,
            created_time=time.time(),
            modified_time=time.time(),
            owner=owner,
            permissions="private",
            storage_class="STANDARD",
            extended_attributes={}
        )
        
        # 上传数据
        if not await self.data_service.upload_object(bucket_name, object_key, data, metadata):
            print("数据上传失败")
            return False
        
        # 存储元数据
        if not await self.metadata_service.put_object_metadata(bucket_name, object_key, metadata):
            print("元数据存储失败")
            # 如果元数据存储失败，应该删除已上传的数据
            await self.data_service.delete_object(bucket_name, object_key)
            return False
        
        return True
    
    async def get_object(self, bucket_name: str, object_key: str) -> Optional[bytes]:
        """获取对象"""
        # 首先检查元数据
        metadata = await self.metadata_service.get_object_metadata(bucket_name, object_key)
        if metadata is None:
            print("对象不存在")
            return None
        
        # 下载数据
        data = await self.data_service.download_object(bucket_name, object_key)
        if data is None:
            print("数据下载失败")
            return None
        
        return data
    
    async def delete_object(self, bucket_name: str, object_key: str) -> bool:
        """删除对象"""
        # 先删除数据
        if not await self.data_service.delete_object(bucket_name, object_key):
            print("数据删除失败")
            return False
        
        # 再删除元数据
        if not await self.metadata_service.delete_object_metadata(bucket_name, object_key):
            print("元数据删除失败")
            return False
        
        return True
    
    async def list_objects(self, bucket_name: str, prefix: str = "") -> List[str]:
        """列出对象"""
        objects = await self.metadata_service.list_objects(bucket_name, prefix)
        return [obj.key for obj in objects]
    
    async def close(self):
        """关闭服务"""
        await self.metadata_service.close_session()
        await self.data_service.close_session()

# 使用示例
async def test_separated_storage_system():
    # 创建分离式存储系统
    storage_system = SeparatedStorageSystem(
        "http://metadata-service:8080",
        "http://data-service:8081"
    )
    
    # 创建存储桶
    print("创建存储桶...")
    await storage_system.metadata_service.create_bucket("my-bucket", "user1")
    
    # 存储对象
    print("存储对象...")
    data1 = b"Hello, World! This is test data 1."
    result1 = await storage_system.put_object("my-bucket", "test/file1.txt", data1, "text/plain")
    print(f"存储对象1结果: {result1}")
    
    data2 = b"Another test file with different content."
    result2 = await storage_system.put_object("my-bucket", "test/file2.txt", data2, "text/plain")
    print(f"存储对象2结果: {result2}")
    
    # 列出对象
    print("列出对象...")
    objects = await storage_system.list_objects("my-bucket", "test/")
    print(f"对象列表: {objects}")
    
    # 获取对象
    print("获取对象...")
    retrieved_data = await storage_system.get_object("my-bucket", "test/file1.txt")
    if retrieved_data:
        print(f"获取到数据: {retrieved_data.decode()}")
    else:
        print("获取对象失败")
    
    # 删除对象
    print("删除对象...")
    delete_result = await storage_system.delete_object("my-bucket", "test/file1.txt")
    print(f"删除对象结果: {delete_result}")
    
    # 再次列出对象
    print("再次列出对象...")
    objects = await storage_system.list_objects("my-bucket", "test/")
    print(f"对象列表: {objects}")
    
    # 关闭服务
    await storage_system.close()

# 运行测试
if __name__ == "__main__":
    # 注意：这需要实际的元数据服务和数据服务运行
    # asyncio.run(test_separated_storage_system())
    print("分离式存储系统示例代码已准备就绪")