异步文件传输任务数≥9时MD5不一致问题排查与解决

1. 并发写入的顺序失控：当异步任务数超过阈值（如你测试的8个），asyncio的任务调度会更频繁，不同任务对应的文件块可能因调度优先级或网络传输延迟，导致接收端收到块的顺序与原文件块顺序不一致。若接收端收到块就直接追加写入文件，会出现块乱序——总大小与原文件一致，但内容顺序被打乱，MD5自然不匹配。
2. aiofiles线程池调度冲突：aiofiles底层通过线程池实现异步文件I/O，默认线程池大小与CPU核心数相关。当任务数超过线程池容量时，额外任务进入等待队列，线程调度可能导致不同块的写入操作交叉执行，破坏文件内容的顺序性。

方案1：基于块编号的有序写入（推荐） #

• 发送端：将文件分割为固定大小的块，每个块传输前添加块编号、总块数的二进制头部（比如前8字节存编号，后8字节存总块数）。
• 接收端：
  • 创建字典或列表缓存已接收的块，以块编号为键。
  • 每收到一个块，先解析头部编号，将块内容存入缓存。
  • 当缓存集齐所有块后，按编号从小到大的顺序，依次写入最终文件。
• 优势：彻底规避乱序问题，即使网络传输顺序混乱也不影响最终结果。

方案2：强制写入顺序的同步控制 #

• 使用asyncio.Lock()对文件写入操作加锁，确保同一时间只有一个任务在写入文件：

  import asyncio
  import aiofiles
  
  write_lock = asyncio.Lock()
  
  async def write_block(file_path, block_data):
      async with write_lock:
          async with aiofiles.open(file_path, 'ab') as f:
              await f.write(block_data)
  

• 注意：该方式会降低并发写入效率，因为写入操作变为串行，但能严格保证顺序正确。

方案3：限制并发任务数与线程池大小 #

• 用asyncio.Semaphore限制同时运行的传输任务数，比如维持在之前正常的8个，避免任务过多导致调度混乱。
• 手动调整aiofiles的线程池大小，匹配任务数量以减少调度冲突：

  import concurrent.futures
  import aiofiles
  
  executor = concurrent.futures.ThreadPoolExecutor(max_workers=16)
  aiofiles.set_default_executor(executor)
  

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