Python - 使用线程或队列遍历调用函数的for循环

Question

我对python相当陌生，正在制作一个脚本，允许将其他程序的点云数据导入Autodesk Maya。我的脚本运行良好，但我想要做的是让它更快。我有一个循环遍历编号文件的列表。即datafile001.txt，datafile002.txt等。我想知道的是，是否有办法让它一次执行多个，可能使用线程或队列？下面我的代码我一直在努力：

 def threadedFuntion(args): 
     if len(sourceFiles) > 3: 
      for count, item in enumerate(sourceFiles): 
        t1=Thread(target=convertPcToPdc,args=(sourceFiles[filenumber1], particlesName, startframe, endframe, pdcIncrements, outputDirectory, variableFolder, acceptableArrayforms, dataType)) 
        t1.start() 
        t2=Thread(target=convertPcToPdc,args=(sourceFiles[filenumber2], particlesName, startframe, endframe, pdcIncrements, outputDirectory, variableFolder, acceptableArrayforms, dataType)) 
        t2.start() 
        t3=Thread(target=convertPcToPdc,args=(sourceFiles[filenumber3], particlesName, startframe, endframe, pdcIncrements, outputDirectory, variableFolder, acceptableArrayforms, dataType)) 
        t3.start() 
        t4=Thread(target=convertPcToPdc,args=(sourceFiles[filenumber4], particlesName, startframe, endframe, pdcIncrements, outputDirectory, variableFolder, acceptableArrayforms, dataType)) 
        t4.start() 

这显然是有很多原因不能正常工作，首先它只是将创建4个线程，我希望能够给一个选项，更或更少。其次它错误，因为它试图重用一个线程？就像我说的，我对python很陌生，而且头脑微微一点，我在这里看过几篇文章，但无法让其工作得很好。我认为队列可能是我需要的东西，但无法弄清楚，我尝试了条件语句和加入语句，但再次无法得到我想要的。

我想更具体什么我想实现的是，该函数是通过文本文件读取，检索coords，然后将它们导出为二进制文件供maya读取。这些文本文件中的一个具有5-10百万个x，y，z坐标，这很常见，这需要相当长的时间。大约需要30分钟 - 1小时的时间在一台漂亮的兽人电脑上做1个文件，任务管理器说python只使用12％的处理器和大约1％的RAM，所以如果我可以同时做多个这样的文件，更多文件的速度要快很多。我不认为多线程/排队for循环很难，但我已经迷路了，并且尝试了一周左右的失败解决方案。

谢谢大家的帮助，我真的很感激，并认为这个网站是惊人的。这是我的第一篇文章，但我觉得我只是从阅读这里完全学会了Python。

Answer 1

子类threading.Thread作为run（）的一部分放在那个类中。

import threading 
import time 
import random 

class Worker(threading.Thread): 
    def __init__(self, srcfile, printlock,**kwargs): 
     super(Worker,self).__init__(**kwargs) 
     self.srcfile = srcfile 
     self.lock = printlock # so threads don't step on each other's prints 

    def run(self): 
     with self.lock: 
      print("starting %s on %s" % (self.ident,self.srcfile)) 
     # do whatever you need to, return when done 
     # example, sleep for a random interval up to 10 seconds 
     time.sleep(random.random()*10) 
     with self.lock: 
      print("%s done" % self.ident) 


def threadme(srcfiles): 
    printlock = threading.Lock() 
    threadpool = [] 
    for file in srcfiles: 
     threadpool.append(Worker(file,printlock)) 

    for thr in threadpool: 
     thr.start() 

    # this loop will block until all threads are done 
    # (however it won't necessarily first join those that are done first) 
    for thr in threadpool: 
     thr.join() 

    print("all threads are done") 

if __name__ == "__main__": 
    threadme(["abc","def","ghi"]) 

按照要求，以限制线程数，使用以下命令：

def threadme(infiles,threadlimit=None,timeout=0.01): 
    assert threadlimit is None or threadlimit > 0, \ 
      "need at least one thread"; 
    printlock = threading.Lock() 
    srcfiles = list(infiles) 
    threadpool = [] 

    # keep going while work to do or being done 
    while srcfiles or threadpool: 

     # while there's room, remove source files 
     # and add to the pool 
     while srcfiles and \ 
      (threadlimit is None \ 
      or len(threadpool) < threadlimit): 
      file = srcfiles.pop() 
      wrkr = Worker(file,printlock) 
      wrkr.start() 
      threadpool.append(wrkr) 

     # remove completed threads from the pool 
     for thr in threadpool: 
      thr.join(timeout=timeout) 
      if not thr.is_alive(): 
       threadpool.remove(thr) 

    print("all threads are done") 

if __name__ == "__main__": 
    for lim in (1,2,3,4): 
     print("--- Running with thread limit %i ---" % lim) 
     threadme(("abc","def","ghi"),threadlimit=lim) 

注意，这实际上在相反的过程中源（由于列表弹出（））。如果您需要按顺序完成它们，请在某处反转列表，或者使用deque和popleft（）。

Answer 2

我会建议使用mrjob为此。

乔先生是一个python执行map reduce。

下面是先生的工作代码做一个多线程的字数超过了很多的文本文件：

from mrjob.job import MRJob 

class MRWordCounter(MRJob): 
    def get_words(self, key, line): 
     for word in line.split(): 
      yield word, 1 

    def sum_words(self, word, occurrences): 
     yield word, sum(occurrences) 

    def steps(self): 
     return [self.mr(self.get_words, self.sum_words),] 

if __name__ == '__main__': 
    MRWordCounter.run() 

此代码的所有文件映射并行（计算每个文件的话），然后降低各种计数成一个单一的总字数。