Python多线程文件处理

Question

我有很少的文件驻留在服务器上,我试图实现一个多线程进程来提高性能,我读了一个教程,但实现它的问题很少,

这是文件,

filelistread = ['h:\\file1.txt', \
                'h:\\file2.txt', \
                'h:\\file3.txt', \
                'h:\\file4.txt']

filelistwrte = ['h:\\file1-out.txt','h:\\file2-out.txt','h:\\file3-out.txt','h:\\file4-out.txt']


def workermethod(inpfile, outfile):
    f1 = open(inpfile,'r')
    f2 = open(outfile,'w')
    x = f1.readlines()
    for each in x:
        f2.write(each)
    f1.close()
    f2.close()

如何使用线程类和队列实现？

我从下面的类开始,但不知道如何将inpfile和outputfile传递给run方法..任何输入都是值得赞赏的

class ThreadUrl(threading.Thread):
    def __init__(self,queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):
        while True:
            item = self.queue.get()

Answer 1

你正在混合两种不同的解决方案.

如果要为每个文件创建专用的工作线程,则不需要任何队列.如果你想创建一个线程池和文件的队列,你不希望通过inpfile和outfile该run方法; 你想把它们放在队列中的每个作业中.

你如何在两者之间做出选择？好了,第一显然是简单的,但如果你有,比如说,1000页要复制的文件,你最终会创造1000个线程,为更多的线程比你想创建,并远远超过线程并行拷贝数操作系统将能够处理.线程池允许您创建8个线程,并在队列中放置1000个作业,并且它们将根据需要分配给线程,因此一次运行8个作业.

让我们从解决方案1开始,解决方案1是每个文件的专用工作线程.

首先,如果你没有与子类化结婚Thread,那么这里真的没有理由这样做.您可以将target函数和args元组传递给默认构造函数,然后该run方法将target(*args)完全按照您的意愿执行.所以:

t = threading.Thread(target=workermethod, args=(inpfile, outfile))

这就是你所需要的.当每个线程运行时,它将调用workermethod(inpfile, outfile)然后退出.

但是,如果由于某种原因确实想要子类Thread,则可以.您可以在构造时传递inpfile和outfile,并且您的run方法将被workermethod修改为使用self.inpfile而self.outfile不是获取参数.像这样:

class ThreadUrl(threading.Thread):
    def __init__(self, inpfile, outfile):
        threading.Thread.__init__(self)
        self.inpfile, self.outfile = inpfile, outfile

    def run(self):
        f1 = open(self.inpfile,'r')
        f2 = open(self.outfile,'w')
        x = f1.readlines()
        for each in x:
            f2.write(each)
        f1.close()
        f2.close()

无论哪种方式,我建议使用with语句,而不是明确的open和close,和摆脱的readlines(不必要地将整个文件读取到内存中),除非你需要处理很老版本的Python:

    def run(self):
        with open(self.inpfile,'r') as f1, open(self.outfile,'w') as f2:
            for line in f1:
                f2.write(line)

现在,解决方案2:线程池和队列.

同样,你不需要这里的子类; 两种做事方式之间的差异与解决方案1中的差异相同.但是坚持你已经开始的子类设计,你需要这样的东西:

class ThreadUrl(threading.Thread):
    def __init__(self,queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def run(self):
        while True:
            inpfile, outfile = self.queue.get()
            workermethod(inpfile, outfile)

然后通过将一个单独传递queue给所有线程来启动线程:

q = queue.Queue
threadpool = [ThreadUrl(q) for i in range(poolsize)]

并提交这样的工作:

q.put((inpfile, outfile))

如果您要使用线程池进行认真的工作,您可能需要考虑使用强大,灵活,简单和优化的实现,而不是自己编写代码.例如,您可能希望能够取消作业,很好地关闭队列,加入整个池而不是逐个加入线程,进行批处理或智能负载平衡等.

如果您使用的是Python 3,则应该查看标准库ThreadPoolExecutor.如果您遇到Python 2,或者无法弄清楚Future,那么您可能需要查看模块中ThreadPool隐藏的类multiprocessing.这两者都具有从多线程切换到多处理的优势(例如,事实证明,您需要与IO一起进行并行化的一些CPU限制工作)是微不足道的.你也可以搜索PyPI,你会发现其他多个好的实现.

作为旁注,您不希望调用队列queue,因为这会影响模块名称.而且,有一些东西被称为workermethod实际上是一个自由函数而不是一个方法,这有点令人困惑.

最后,如果你所做的只是复制文件,你可能不想在文本模式下阅读,或者逐行阅读.事实上,你可能根本不想自己实现它; 只需使用相应的复制功能shutil.您可以非常轻松地使用上述任何方法.例如,而不是这样:

t = threading.Thread(target=workermethod, args=(inpfile, outfile))

做这个:

t = threading.Thread(target=shutil.copyfile, args=(inpfile, outfile))

事实上,您的整个程序看起来可以替换为:

threads = [threading.Thread(target=shutil.copyfile, args=(inpfile, outfile))
           for (inpfile, outfile) in zip(filelistread, filelistwrte)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()