Python多进程拷贝文件并显示进度：原理、实现与性能分析

程一诺

发布时间：2026-10-14 | 4人浏览过

原创

本文介绍使用Python实现多进程拷贝文件夹并实时显示百分比进度的方法。通过对比单进程与多进程在1核CPU环境下的执行耗时，分析任务切换开销对性能的影响，并提供基于shutil模块的完整代码示例供参考。

Python多进程文件拷贝功能概述

本文分享一个使用Python实现的多进程文件拷贝功能示例，该功能支持实时显示拷贝进度的百分比。内容仅供学习参考。

环境检查与实现逻辑

在CentOS 7系统中，可以通过特定命令查看CPU的核数。

查看CPU核数命令


# 总核数 = 物理CPU个数 X 每颗物理CPU的核数
# 总逻辑CPU数 = 物理CPU个数 X 每颗物理CPU的核数 X 超线程数
# 查看物理CPU个数
cat /proc/cpuinfo| grep "physical id"| sort| uniq| wc -l
# 查看每个物理CPU中core的个数(即核数)
cat /proc/cpuinfo| grep "cpu cores"| uniq
# 查看逻辑CPU的个数
cat /proc/cpuinfo| grep "processor"| wc -l

本示例中的文件拷贝逻辑虽然基础，但实现过程相对复杂，主要涉及多进程管理与进度同步。

第一步，根据输入的源文件夹名称，程序会自动创建一个新的目标文件夹用于存储拷贝的文件。

第二步，程序读取源文件夹内的所有文件内容，并将这些内容写入到目标文件夹中同名文件里。

以下是具体的Python代码实现：

基础多进程拷贝代码


# coding=utf-8
from multiprocessing import Pool, Manager
import os
import traceback
import time
def copyFileTask(name, oldFolderName, newFolderName, queue):
  """copy文件"""
  fr = open(oldFolderName+"/"+name)
  fw = open(newFolderName+"/"+name, "w")
  content = fr.read()
  fw.write(content)
  fr.close()
  fw.close()
  # 将copy完成的文件名称put进队列
  queue.put(name)
def main():
  try:
    # 0. 获取要copy的文件夹
    # oldFolderName = input("请输入文件夹的名字：") # python3
    oldFolderName = raw_input("请输入文件夹的名字：") # python2
    # 1. 创建一个新的文件夹
    print 'mkdir new file ------'
    newFolderName = str(oldFolderName) + "-copy"
    # print(newFolderName)
    os.mkdir(newFolderName)
    # 2. 获取old文件夹中的所有的文件名字
    fileNames = os.listdir(oldFolderName)
    # print(fileNames)
    # 3. 使用多进程的方式copy 原文件夹中的所有文件到新的文件夹中
    pool = Pool(5)
    # 创建一个队列
    queue = Manager().Queue()
    for name in fileNames:
      pool.apply_async(copyFileTask, args=(name, oldFolderName, newFolderName, queue))
    pool.close() # 关闭进程池，不再接受请求
    pool.join() # 等待所有的子进程结束
    num = 0
    # 需要copy的文件总数
    allNum = len(fileNames)
    while num < allNum:
      # 收数据
      queue.get()
      num += 1
      copyRate = float(num) / allNum
      # 打印copy的进度
      time.sleep(0.1) # 容易看出来打印百分比的变化，延长百分比更新的时间
      print "\rcopy的进度是:%.2f%%" % (copyRate*100),
    print "\ncopy success ------"
  except:
    traceback.print_exc()
if __name__ == "__main__":
  main()

下面演示如何使用Python标准库中的shutil模块来执行拷贝操作。

基于shutil模块的多进程实现


# coding=utf-8
from multiprocessing import Pool, Manager
import os
import traceback
import time
import shutil
def copyFileTask(name, old_path, new_path, queue):
  """copy文件"""
  src_path = os.path.join(old_path, name)
  dst_path = os.path.join(new_path, name)
  shutil.copy(src_path, dst_path) # 拷贝文件
  # shitil.move(src_path, dst_path) # 移动文件
  # 将copy完成的文件名称put进队列
  queue.put(name)
def main():
  try:
    # 0. 获取要copy的文件夹
    oldFolderName = raw_input("请输入文件夹的名字：")
    print '要拷贝的源文件夹： ', oldFolderName
    # 1. 创建一个新的文件夹
    newFolderName = raw_input("请输入文件夹的名字：")
    print '目标文件夹：', newFolderName
    # 获取当前所在的路径
    now_path = os.getcwd()
    print "当前所在路径： ", now_path
    old_path = os.path.join(now_path, oldFolderName)
    print '源文件夹路径： ', old_path
    new_path = os.path.join(now_path, newFolderName)
    print '目标文件夹路径： ', new_path
    if not os.path.exists(new_path):
      os.mkdir(new_path)
    # 3. 使用多进程的方式copy 原文件夹中的所有文件到新的文件夹中
    pool = Pool(5) # 创建5条进程
    queue = Manager().Queue() # 创建一个队列
    # 遍历源文件夹
    for root, dirs, files in os.walk(oldFolderName):
      """
      root 所指的是当前正在遍历的这个文件夹的本身的地址
      dirs 是一个 list ，内容是该文件夹中所有的目录的名字(不包括子目录)
      files 同样是 list , 内容是该文件夹中所有的文件(不包括子目录)
      """
      print '源文件夹下的文件数量:：', len(files)
      s_t = time.time()
      for file in files:
        pool.apply_async(copyFileTask, args=(file, old_path, new_path, queue)) # 向进程池中添加任务
      pool.close() # 关闭进程池，不再接受请求
      pool.join() # 等待所有的子进程结束
      print '耗时：{} 秒'.format(time.time() - s_t)
      num = 0
      # 需要copy的文件总数
      allNum = len(files)
      print 'len(fileNames) = ', allNum
      while True:
        # 收数据
        # print '收数据 ------'
        queue.get()
        # print 'queue.get() ------', queue.get()
        num += 1
        copyRate = float(num) / allNum
        # 打印copy的进度
        print "\rcopy的进度是:%.2f%%" % (copyRate*100),
        if num == allNum:
          break
    print "\ncopy success ------"
  except:
    traceback.print_exc()
if __name__ == "__main__":
  main()

运行结果如下：由于测试环境的CPU核数为1核，在多进程模式下，操作系统需要进行频繁的任务轮询和上下文切换，这导致了额外的时间消耗，使得多进程版本的执行时间反而比单进程更长。

[root@centos7 mnt]# python copy_file_test.py 请输入文件夹的名字：hdfs 要拷贝的源文件夹： hdfs 请输入文件夹的名字：test01 目标文件夹： test01 当前所在路径： /mnt 源文件夹路径： /mnt/hdfs 目标文件夹路径： /mnt/test01 源文件夹下的文件数量:： 8165 耗时：253.655323982 秒 len(fileNames) = 8165 copy的进度是:100.00% copy success ------

作为对比，以下是单进程拷贝文件的实现方式：

单进程拷贝代码实现


# coding=utf-8
from Queue import Queue
import os
import traceback
import time
import shutil
def copyFileTask(name, old_path, new_path, queue):
  """copy文件"""
  src_path = os.path.join(old_path, name)
  dst_path = os.path.join(new_path, name)
  shutil.copy(src_path, dst_path) # 拷贝文件
  # shitil.move(src_path, dst_path) # 移动文件
  # 将copy完成的文件名称put进队列
  queue.put(name)
def main():
  try:
    # 0. 获取要copy的文件夹
    oldFolderName = raw_input("请输入文件夹的名字：")
    print '要拷贝的源文件夹： ', oldFolderName
    # 1. 创建一个新的文件夹
    newFolderName = raw_input("请输入文件夹的名字：")
    print '目标文件夹：', newFolderName
    # 获取当前所在的路径
    now_path = os.getcwd()
    print "当前所在路径： ", now_path
    old_path = os.path.join(now_path, oldFolderName)
    print '源文件夹路径： ', old_path
    new_path = os.path.join(now_path, newFolderName)
    print '目标文件夹路径： ', new_path
    if not os.path.exists(new_path):
      os.mkdir(new_path)
    queue = Queue() # 创建一个队列
    # 遍历源文件夹
    for root, dirs, files in os.walk(oldFolderName):
      """
      root 所指的是当前正在遍历的这个文件夹的本身的地址
      dirs 是一个 list ，内容是该文件夹中所有的目录的名字(不包括子目录)
      files 同样是 list , 内容是该文件夹中所有的文件(不包括子目录)
      """
      print '源文件夹下的文件数量:：', len(files)
      s_t = time.time()
      # 调用函数拷贝文件
      for file in files:
        copyFileTask(file, old_path, new_path, queue)
      print '耗时：{} 秒'.format(time.time() - s_t)
      num = 0
      # 需要copy的文件总数
      allNum = len(files)
      print 'len(fileNames) = ', allNum
      while True:
        # 收数据
        # print '收数据 ------'
        queue.get()
        # print 'queue.get() ------', queue.get()
        num += 1
        copyRate = float(num) / allNum
        # 打印copy的进度
        print "\rcopy的进度是:%.2f%%" % (copyRate*100),
        if num == allNum:
          break
    print "\ncopy success ------"
  except:
    traceback.print_exc()
if __name__ == "__main__":
  main()

单进程模式的运行输出如下：

[root@centos7 mnt]# python copy_file_test_dan.py 请输入文件夹的名字：hdfs 要拷贝的源文件夹： hdfs 请输入文件夹的名字：test02 目标文件夹： test02 当前所在路径： /mnt 源文件夹路径： /mnt/hdfs 目标文件夹路径： /mnt/test02 源文件夹下的文件数量:： 8165 耗时：122.284090996 秒 len(fileNames) = 8165 copy的进度是:100.00% copy success ------

本次测试使用的虚拟机CPU核数为1。

具体的CPU信息查看结果如下：

测试环境CPU配置信息


[root@centos7 mnt]# cat /proc/cpuinfo| grep "physical id"| sort| uniq| wc -l
1
[root@centos7 mnt]# cat /proc/cpuinfo| grep "cpu cores"| uniq
cpu cores : 1
[root@centos7 mnt]# cat /proc/cpuinfo| grep "processor"| wc -l
1

如果您对Python的文件与目录操作、数据结构、算法或字符串处理感兴趣，建议进一步查阅本站相关的《Python文件与目录操作技巧汇总》、《Python文本文件操作技巧汇总》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》以及《Python入门与进阶经典教程》等专题文章。

希望上述关于Python程序设计的示例和说明能对读者有所帮助。