Python 实战使用 进程池 多进程 copy文件

仅供学习参考，转载请注明出处

多进程copy文件

为了更加号的理解多进程，编写一个批量copy文件的案例。

环境：centos7
需求：读取一个文件夹中的多个文件，然后写入到另一个文件夹中

编写的前置知识点

掌握​​os​​遍历读取文件夹的相关命令：

In [1]: import os # 导入os模块

In [5]: os.getcwd() # 获取当前文件路径
Out[5]: '/work'

In [6]:

In [6]: os.listdir('/work') # 获取 /work 目录下的文件
Out[6]: ['udp', 'tcp', 'tcp_download', 'many_task', 'process']

In [7]:

In [7]: os.listdir('../') # 获取 ../ 目录下的文件
Out[7]:
['boot',
'dev',
'proc',
'run',
'sys',
'etc',
'root',
'var',
'tmp',
'usr',
'bin',
'sbin',
'lib',
'lib64',
'home',
'media',
'mnt',
'opt',
'srv',
'dump.rdb',
'work']

In [8]:

In [9]: os.mkdir('copy_dir') # 创建copy_dir的文件夹

In [10]:

In [10]: ls
copy_dir/ many_task/ process/ tcp/ tcp_download/ udp/

In [11]:

那么从上面ipython的基本操作中，可以使用os模块获取需要拷贝的文件目录下的文件（不考虑还有子目录）为list数组。
然后遍历list数组来拷贝数据。

V1.0 首先获取需要复制文件夹的文件以及创建目标文件夹

[root@server01 work]# vim copy_file.py

#coding=utf-8

import os

def main():
# 获取当前文件路径
basedir = os.getcwd()
print("当前文件夹路径：%s" % basedir)

# 设置需要复制的文件夹
src_dir = basedir + "/" + "many_task"
print("复制的源文件夹路径：%s" % src_dir)

# 获取源文件夹路径下的所有文件（第一层）
src_files_list = os.listdir(src_dir)
print("打印源文件下的所有文件：%s" % src_files_list)

# 创建复制到的目标文件夹目录
dst_dir = src_dir + "[copy]"

os.mkdir(dst_dir)

if __name__=="__main__":
main()

执行如下：

[root@server01 work]# ls
copy_dir copy_file.py many_task process tcp tcp_download udp
[root@server01 work]#
[root@server01 work]# python copy_file.py
当前文件夹路径：/work
复制的源文件夹路径：/work/many_task
打印源文件下的所有文件：['test.py', 'test2.py', 'test3.py', 'test4.py', 'test5.py', 'test6.py', 'test7.py', 'test8.py', 'test9.py', 'test10.py']
[root@server01 work]#
[root@server01 work]# ls
copy_dir copy_file.py many_task many_task[copy] process tcp tcp_download udp
[root@server01 work]#

有了目标文件夹之后，那么下一步就可以开始写复制文件的步骤了。

V1.1 - 编写根据获取的文件list，将文件复制到目标文件夹中

#coding=utf-8

import os

def main():
# 获取当前文件路径
basedir = os.getcwd()
print("当前文件夹路径：%s" % basedir)

# 设置需要复制的文件夹
src_dir = basedir + "/" + "many_task"
print("复制的源文件夹路径：%s" % src_dir)

# 获取源文件夹路径下的所有文件（第一层）
src_files_list = os.listdir(src_dir)
print("打印源文件下的所有文件：%s" % src_files_list)

# 创建复制到的目标文件夹目录
dst_dir = src_dir + "[copy]"

try:
os.mkdir(dst_dir)
except OSError:
print("文件夹已创建")

# 遍历需要复制的文件
for src_file in src_files_list:
print("打印需要拷贝的源文件：%s" % src_file)
# 复制文件至目标文件夹中
file_read = open(src_dir + "/" + src_file,"rb")
file_write = open(dst_dir + "/" + src_file,"wb")
while True:
content = file_read.read(1024) # 读取文件内容
if content:
file_write.write(content)
else:
break
file_read.close()
file_write.close()

if __name__=="__main__":
main()

执行如下：

#coding=utf-8

import os

def main():
# 获取当前文件路径
basedir = os.getcwd()
print("当前文件夹路径：%s" % basedir)

# 设置需要复制的文件夹
src_dir = basedir + "/" + "many_task"
print("复制的源文件夹路径：%s" % src_dir)

# 获取源文件夹路径下的所有文件（第一层）
src_files_list = os.listdir(src_dir)
print("打印源文件下的所有文件：%s" % src_files_list)

# 创建复制到的目标文件夹目录
dst_dir = src_dir + "[copy]"

try:
os.mkdir(dst_dir)
except OSError:
print("文件夹已创建")

# 遍历需要复制的文件
for src_file in src_files_list:
print("打印需要拷贝的源文件：%s" % src_file)
# 复制文件至目标文件夹中
file_read = open(src_dir + "/" + src_file,"rb")
file_write = open(dst_dir + "/" + src_file,"wb")
while True:
content = file_read.read(1024) # 读取文件内容
if content:
file_write.write(content)
else:
break
file_read.close()
file_write.close()

if __name__=="__main__":
main()

从执行结果来看，已经实现了复制文件的目标了。但是如果文件数量非常多，文件非常大。
这样循环复制的话会效率较低，那么下面就要考虑如何多进程执行这个拷贝的动作了。

V2.0 - 多进程拷贝文件

那么，需要分析需要拆分下面的几个步骤：

• 将拷贝的动作写成一个方法，后续可以用来进程调用
• 创建一个进程池，用于管理进程的并发数量
• 创建一个进程池的队列，用于打印已经完成拷贝的文件名称

实现代码如下：

#coding=utf-8

import multiprocessing
import os

def copy_file(queue,src_dir,src_file,dst_dir):
# 复制文件至目标文件夹中
file_read = open(src_dir + "/" + src_file,"rb")
file_write = open(dst_dir + "/" + src_file,"wb")
while True:
content = file_read.read(1024) # 读取文件内容
if content:
file_write.write(content)
else:
break
file_read.close()
file_write.close()

queue.put(src_file) # 将执行拷贝完毕的文件名放入队列中

def main():
# 获取当前文件路径
basedir = os.getcwd()
print("当前文件夹路径：%s" % basedir)

# 设置需要复制的文件夹
src_dir = basedir + "/" + "many_task"
print("复制的源文件夹路径：%s" % src_dir)

# 获取源文件夹路径下的所有文件（第一层）
src_files_list = os.listdir(src_dir)
print("打印源文件下的所有文件：%s" % src_files_list)

# 创建复制到的目标文件夹目录
dst_dir = src_dir + "[copy]"

try:
os.mkdir(dst_dir)
except OSError:
print("文件夹已创建")

# 创建进程池
pool = multiprocessing.Pool(3) # 创建三个进程的进程池

# 创建队列
queue = multiprocessing.Manager().Queue()

# 遍历需要复制的文件
for src_file in src_files_list:
print("打印需要拷贝的源文件：%s" % src_file)
#copy_file(src_dir,src_file,dst_dir)

# 向进程池添加任务
pool.apply_async(copy_file,args=(queue,src_dir,src_file,dst_dir))

# 关闭进程池
pool.close()
pool.join()

# 从队列中获取拷贝完毕的文件名
while True:
file_name = queue.get()
print("已经拷贝完毕的文件:%s" % file_name)

# 判断文件全部打印完毕，退出循环
if file_name in src_files_list:
src_files_list.remove(file_name)

print("当前源文件list的大小：%d" % len(src_files_list))
if len(src_files_list) == 0:
break

if __name__=="__main__":
main()

执行如下：

[root@server01 work]# python copy_file.py
当前文件夹路径：/work
复制的源文件夹路径：/work/many_task
打印源文件下的所有文件：['test.py', 'test2.py', 'test3.py', 'test4.py', 'test5.py', 'test6.py', 'test7.py', 'test8.py', 'test9.py', 'test10.py']
打印需要拷贝的源文件：test.py
打印需要拷贝的源文件：test2.py
打印需要拷贝的源文件：test3.py
打印需要拷贝的源文件：test4.py
打印需要拷贝的源文件：test5.py
打印需要拷贝的源文件：test6.py
打印需要拷贝的源文件：test7.py
打印需要拷贝的源文件：test8.py
打印需要拷贝的源文件：test9.py
打印需要拷贝的源文件：test10.py
已经拷贝完毕的文件:test.py
当前源文件list的大小：9
已经拷贝完毕的文件:test2.py
当前源文件list的大小：8
已经拷贝完毕的文件:test3.py
当前源文件list的大小：7
已经拷贝完毕的文件:test4.py
当前源文件list的大小：6
已经拷贝完毕的文件:test5.py
当前源文件list的大小：5
已经拷贝完毕的文件:test6.py
当前源文件list的大小：4
已经拷贝完毕的文件:test7.py
当前源文件list的大小：3
已经拷贝完毕的文件:test8.py
当前源文件list的大小：2
已经拷贝完毕的文件:test9.py
当前源文件list的大小：1
已经拷贝完毕的文件:test10.py
当前源文件list的大小：0
[root@server01 work]#

关注微信公众号，回复【资料】、Python、PHP、JAVA、web，则可获得Python、PHP、JAVA、前端等视频资料。