python多线程文件拷贝

任务类型可以分为计算密集型和IO密集型

计算密集型：特点是要进行大量的计算，消耗CPU资源，比如计算圆周率、对视频进行高清解码等等，全靠CPU的运算能力。这种计算密集型任务虽然也可以用多任务完成，但是任务越多，花在任务切换的时间就越多，CPU执行任务的效率就越低，所以，要最高效地利用CPU，计算密集型任务同时进行的数量应当等于CPU的核心数。

IO密集型：涉及到网络、磁盘IO的任务都是IO密集型任务，这类任务的特点是CPU消耗很少，任务的大部分时间都在等待IO操作完成（因为IO的速度远远低于CPU和内存的速度）。对于IO密集型任务，任务越多，CPU效率越高，但也有一个限度。常见的大部分任务都是IO密集型任务，比如Web应用。

摘自：https://www.liaoxuefeng.com/wiki/1016959663602400/1017631469467456

文件拷贝是IO密集型的任务，适合多线程完成，以下代码实现了多线程文件拷贝。修改SRC_PATH（源地址）、DST_PATH（目标地址）两个参数。

 

# coding: utf-8

"""
开启多线程上传文件
python upload.py -h 查看说明文档
"""


import os
import shutil
import time
import logging
import threading
import multiprocessing
from multiprocessing.dummy import Pool as ThreadPool
import argparse

DEFAULT_THREADING = multiprocessing.cpu_count() # 默认线程数


def getsize(filename):
    """Return the size of a file, reported by os.stat()."""
    return os.stat(filename).st_size

def getdirsize(path):
    size = 0
    for root, dirs, files in os.walk(path):
        size += sum([getsize(os.path.join(root, name)) for name in files])
    return size

# 字节bytes转化kbmg
def formatsize(bytes):
    try:
        bytes = float(bytes)
        kb = bytes / 1024
    except Exception as e:
        logging.error(e)
        return "Format Error"

    if kb >= 1024:
        M = kb / 1024
        if M >= 1024:
            G = M / 1024
            return "%.2fG" % (G)
        else:
            return "%.2fM" % (M)
    else:
        return "%.2fkb" % (kb)

def parse_argv():
    parser = argparse.ArgumentParser(description='配置线程个数')
    parser.add_argument('-t', dest='threading', metavar='线程数', action='store',
            type=int, default=DEFAULT_THREADING, help='线程数')
    args = parser.parse_args()
    return args.threading

def copy_file(src, dst):
    """拷贝文件

    :param src: 源地址
    :param dst: 目标地址
    :return: None
    """
    logging.info("正在拷贝文件：%s, 大小: %s" % (src, formatsize(getsize(src))))
    if not os.path.isfile(src):
        logging.error("错误，源地址不是一个文件：%s" % src)
        return
    # Copy src to dst. (cp src dst)
    shutil.copy(src, dst)
    logging.info("文件拷贝完成：%s" % src)

def threading_main(num, src_path, dst_path):
    """开启多个线程
    :param num: 线程数
    :param src_path: 源地址
    :param dst_path: 目标地址
    """
    logging.info("进程pid(%s)，即将开启%s个线程" % (os.getpid(), num))
    pool = ThreadPool(num)  # 开启num个线程
    for root, dirs, files in os.walk(src_path):
        for f in files:
            if f.startswith("."):
                continue
            src = os.path.join(src_path, root, f)
            dst = os.path.abspath(src.replace(src_path, dst_path))
            dst_dir = os.path.dirname(dst)
            if not os.path.isdir(dst_dir):
                os.makedirs(dst_dir)
            pool.apply_async(
                    func=copy_file,
                    args=(src, dst, )
                    )
    pool.close()
    pool.join()


def main(src_path, dst_path):
    t = parse_argv()
    if t <= 0 or t > DEFAULT_THREADING * 8:
        t = DEFAULT_THREADING
    logging.info("需拷贝文件总大小：%s" % formatsize(getdirsize(src_path)))
    threading_main(t, src_path, dst_path)


if __name__ == "__main__":
    SRC_PATH = "" # 源地址
    DST_PATH = "" # 目标地址
    if not os.path.isdir(SRC_PATH):
        logging.error("请填写正确的上传路径！！！")
        exit(1)
    FMT = (
            '[%(levelname)s][%(name)s:%(process)d:%(threadName)s][%(asctime)s]' +
        ': %(message)s')
    logging.basicConfig(level=logging.INFO, format=FMT)
    logging.info("---开始拷贝---")
    start = int(time.time())
    main(SRC_PATH, DST_PATH)
    end = int(time.time())
    logging.info("---拷贝完成---")
    logging.info("用时：%s 秒，%s 分钟" % (end-start, (end-start)//60))