从 MinIO 到 RustFS：简单迁移工具

前言

原因是在2025年2月26日MinIO合并的一个PR中，删除了114736行代码，删除了Web UI的几乎所有核心管理功能，仅保留对象浏览能力。也就是镜像minio/minio:RELEASE.2025-04-08T15-41-24Z 后面都是阉割版，在写这篇文章，最新的MinIO的镜像需要用户自己构建。

因此考虑其他开源的对象存储解决方案，很多人推荐国产存储项目RustFS。作为新兴的存储解决方案，凭借高性能和灵活性逐渐受到关注，所以本次实现的迁移工具需达成以下目标：

• 支持多类型 RustFS 后端（S3 兼容接口、本地文件系统、HTTP API）

• 具备高并发迁移能力，兼顾效率与稳定性

• 完善的错误处理与重试机制，保障数据完整性

• 安全防护（如路径遍历攻击防范）

• 详细的进度跟踪与日志记录

工具架构

工具采用模块化设计，主要分为 5 个核心模块：

1. 基础工具模块：配置加载、日志初始化、路径安全验证等通用功能

2. MinIO 客户端模块：负责从 MinIO 读取对象元数据与数据流

3. RustFS 客户端模块：适配不同类型的 RustFS 后端，提供统一上传接口

4. 迁移核心模块：实现并发调度、进度统计与结果汇总

5. 主程序模块：协调各组件工作，完成端到端迁移流程

核心功能

配置与环境

配置加载采用 YAML 格式，支持默认值填充，提升容错性：

def load_config():
    config_path = os.path.join(get_exe_dir(), "config.yml")
    if not os.path.exists(config_path):
        logger.error(f"配置文件不存在：{config_path}")
        sys.exit(1)

日志系统支持文件与控制台双输出，兼容 Windows 环境的中文显示与权限问题：

def init_logger():
    # 日志目录自适应（优先EXE目录，权限不足时切换到桌面）
    log_dir = Path(get_exe_dir()) / "logs"
    try:
        log_dir.mkdir(exist_ok=True, parents=True)
    except PermissionError:
        log_dir = Path(os.path.expanduser("~")) / "Desktop" / "migrate_logs"
        log_dir.mkdir(exist_ok=True, parents=True)

MinIO客户端

MinIO 客户端实现了对象列表获取与数据流读取功能，特别注意路径安全验证：

class MinioClient:
    def __init__(self, config):
        self.client = Minio(
            endpoint=config["endpoint"],
            access_key=config["access_key"],
            secret_key=config["secret_key"],
            secure=config["secure"],
        )
        if not self.client.bucket_exists(self.bucket):
            raise ValueError(f"Minio桶不存在：{self.bucket}")

RustFS 客户端

多后端适配

class RustfsClient:
    def __init__(self, config, migrate_config):
        self.type = config["type"].lower()
        if self.type == "s3":
            self._init_s3_client(config["s3"])
        elif self.type == "local":
            self._init_local_client(config["local"])
        elif self.type == "http":
            self._init_http_client(config["http"])

并发调度与进度跟踪

def main():
    # 列出待迁移对象
    obj_list = list(minio_client.list_objects_with_size())
    total = len(obj_list)
    logger.info(f"共发现 {total} 个待迁移文件")

    # 动态调整并发数（大文件(>100MB)较多时降低并发）
    large_file_threshold = 100 * 1024 * 1024
    large_file_count = sum(1 for _, size in obj_list if size > large_file_threshold)
    adjusted_concurrency = max(1, migrate_config["concurrency"] - large_file_count // 5)

    # 并发迁移
    with ThreadPoolExecutor(max_workers=adjusted_concurrency) as executor:
        future_to_obj = {
            executor.submit(migrate_single_object, minio_client, rustfs_client, obj_name, size): (obj_name, size)
            for obj_name, size in obj_list
        }

使用指南

资源下载

修改配置

解压migrate.zip ，修改config.yml配置，最小修改如下。

minio:
  endpoint: "IP:9000"
  access_key: "minio_access_key"
  secret_key: "minio_secret_key"
  bucket: "source-bucket"
rustfs:
  type: "s3"
  s3:
    endpoint: "IP:9000"
    access_key: "rustfs_access_key"
    secret_key: "rustfs_secret_key"
    bucket: "target-bucket"

运行工具

直接执行可执行文件或通过 Python 运行migrate_core.py(评论发送：源码)

查看结果

通过控制台输出或logs目录下的日志文件查看迁移详情

其他扩展

实际应用中，可根据需求进一步扩展：

• 增加增量迁移支持（基于文件修改时间或 ETag）

• 实现断点续传功能

• 添加迁移后的数据校验机制

• 扩展更多存储后端支持（如 OSS、GCS 等）

使用必看

此工具是通过一定并发控制、错误处理与安全防护，仅能够满足中小规模数据迁移的需求，如需大规模附件迁移请寻找更合适的方案。