如何用asyncio实现单FTP客户端向多服务器并行推送游戏文件？

问题核心 #

你的代码无法实现并行推送的关键原因是使用了同步的ftplib库——它的所有IO操作都是阻塞式的，会直接卡住asyncio事件循环，导致多个ftp_push协程无法同时运行。要利用1Gb网卡的带宽优势并行推送，必须改用异步FTP客户端。

解决方案 #

1. 安装异步FTP依赖

首先安装专为异步场景设计的aioftp库：

pip install aioftp

2. 重构为异步实现

以下是修改后的完整代码，支持多Xbox并行推送，且可按需求分组（比如每10台一组）：

#!/usr/bin/env python3
import asyncio
import os
from tqdm import tqdm
import aioftp

game_dir = os.path.expanduser("~/360/Games/halo_3")
ftp_host = os.environ['ftp_host']
ftp_pass = os.environ['ftp_pass']
ftp_users = ["xbox1", "xbox2"]
# 分组推送大小：每10台Xbox为一组
GROUP_SIZE = 10

def get_dir_size(path: str) -> int:
    """返回目录总大小（字节）"""
    if not os.path.isdir(path):
        print(f"Error: {path} 不是有效目录")
        exit(1)
    size = 0
    with os.scandir(path) as entries:
        for entry in entries:
            if entry.is_file():
                size += entry.stat().st_size
            elif entry.is_dir():
                size += get_dir_size(entry.path)
    return size

async def ftp_login(host, user, passwd):
    """异步登录FTP服务器"""
    client = aioftp.Client(host, user=user, password=passwd)
    await client.connect()
    print(f"[{user}] 已连接: {await client.get_welcome()}")
    return client

async def ftp_push(client, local_path, remote_path, progress_bar):
    """异步推送目录到FTP服务器"""
    with os.scandir(local_path) as entries:
        for entry in entries:
            current_remote = os.path.join(remote_path, entry.name)
            if entry.is_file():
                # 异步上传文件，回调更新进度条
                async with open(entry.path, "rb") as f:
                    await client.upload(
                        f,
                        current_remote,
                        callback=lambda sent: progress_bar.update(sent)
                    )
            elif entry.is_dir():
                # 异步创建目录（已存在则忽略）
                await client.make_directory(current_remote, exist_ok=True)
                await client.change_directory(current_remote)
                # 递归推送子目录
                await ftp_push(client, entry.path, current_remote, progress_bar)
                await client.change_directory("..")

async def main():
    total_size = get_dir_size(game_dir)
    # 分批处理Xbox设备，避免同时连接过多
    for i in range(0, len(ftp_users), GROUP_SIZE):
        current_group = ftp_users[i:i+GROUP_SIZE]
        print(f"开始推送第 {i//GROUP_SIZE +1} 组，共 {len(current_group)} 台设备")
        
        # 异步登录当前组的所有FTP客户端
        login_tasks = [ftp_login(ftp_host, user, ftp_pass) for user in current_group]
        clients = await asyncio.gather(*login_tasks)
        
        # 创建每个设备的进度条
        progress_bars = [
            tqdm(
                unit='B', unit_scale=True, leave=False, miniters=1,
                desc=user, total=total_size
            )
            for user in current_group
        ]
        
        try:
            # 并行执行当前组的所有推送任务
            push_tasks = [
                ftp_push(client, game_dir, ".", pbar)
                for client, pbar in zip(clients, progress_bars)
            ]
            await asyncio.gather(*push_tasks)
        finally:
            # 确保关闭所有FTP连接和进度条
            for client in clients:
                await client.quit()
            for pbar in progress_bars:
                pbar.close()
        print(f"第 {i//GROUP_SIZE +1} 组推送完成")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"发生错误: {e}")

关键修改说明 #

1. 异步IO核心：用aioftp.Client替代同步的ftplib.FTP，所有FTP操作（连接、创建目录、上传）都是异步方法，不会阻塞事件循环，多个推送任务可真正并行。
2. 分组控制：通过GROUP_SIZE参数实现分批推送，避免同时连接过多设备导致带宽或服务器资源过载。
3. 进度条安全更新：aioftp.upload的回调函数会在每次发送数据块时触发，确保进度条更新线程安全。
4. 资源清理：在finally块中统一关闭FTP连接和进度条，避免资源泄漏。

内容的提问来源于stack exchange，提问作者aymericpineau