如何实现动态数量的并发线程？

Question

我发起的并发线程做一些东西：

concurrent = 10 
q = Queue(concurrent * 2) 
for j in range(concurrent): 
    t = threading.Thread(target=doWork) 
    t.daemon = True 
    t.start() 
try: 
    # process each line and assign it to an available thread 
    for line in call_file: 
     q.put(line) 
    q.join() 
except KeyboardInterrupt: 
    sys.exit(1) 

同时我有一个不同的线程计数时间：

def printit(): 
    threading.Timer(1.0, printit).start() 
    print current_status 

printit() 

我想增加（或减少）的并发量主流程的线程让我们说每分钟。我可以在时间线程中创建一个时间计数器，并让它每分钟执行一些操作，但是如何更改主进程中并发线程的数量？

是否有可能（如果是的话）如何做到这一点？

Answer 1

这是我的工作人员：

def UpdateProcesses(start,processnumber,CachesThatRequireCalculating,CachesThatAreBeingCalculated,CacheDict,CacheLock,IdleLock,FileDictionary,MetaDataDict,CacheIndexDict): 
NewPool() 
while start[processnumber]: 
    IdleLock.wait() 
    while len(CachesThatRequireCalculating)>0 and start[processnumber] == True: 
     CacheLock.acquire() 
     try: 
      cacheCode = CachesThatRequireCalculating[0] # The list can be empty if an other process takes the last item during the CacheLock 
      CachesThatRequireCalculating.remove(cacheCode) 
      print cacheCode,"starts processing by",processnumber,"process" 
     except: 
      CacheLock.release() 
     else: 
      CacheLock.release() 
      CachesThatAreBeingCalculated.append(cacheCode[:3]) 
      Array,b,f = TIPP.LoadArray(FileDictionary[cacheCode[:2]])#opens the dask array 
      Array = ((Array[:,:,CacheIndexDict[cacheCode[:2]][cacheCode[2]]:CacheIndexDict[cacheCode[:2]][cacheCode[2]+1]].compute()/2.**(MetaDataDict[cacheCode[:2]]["Bit Depth"])*255.).astype(np.uint16)).transpose([1,0,2]) #slices and calculates the array 
      f.close() #close the file 
      if CachesThatAreBeingCalculated.count(cacheCode[:3]) != 0: #if not, this cache is not needed annymore (the cacheCode is removed bij wavelengthchange) 
       CachesThatAreBeingCalculated.remove(cacheCode[:3]) 
       try: #If the first time the object if not aivalable try a second time 
        CacheDict[cacheCode[:3]] = Array 
       except: 
        CacheDict[cacheCode[:3]] = Array 
       print cacheCode,"done processing by",processnumber,"process" 
    if start[processnumber]: 
     IdleLock.clear() 

这是我如何开始他们：

self.ProcessLst = [] #list with all the processes who calculate the caches 
    for processnumber in range(min(NumberOfMaxProcess,self.processes)): 
     self.ProcessTerminateLst.append(True) 
    for processnumber in range(min(NumberOfMaxProcess,self.processes)): 
     self.ProcessLst.append(process.Process(target=Proc.UpdateProcesses,args=(self.ProcessTerminateLst,processnumber,self.CachesThatRequireCalculating,self.CachesThatAreBeingCalculated,self.CacheDict,self.CacheLock,self.IdleLock,self.FileDictionary,self.MetaDataDict,self.CacheIndexDict,))) 
     self.ProcessLst[-1].daemon = True 
     self.ProcessLst[-1].start() 

我关闭它们是这样的：

for i in range(len(self.ProcessLst)): #For both while loops in the processes self.ProcessTerminateLst[i] must be True. So or the process is now ready to be terminad or is still in idle mode. 
     self.ProcessTerminateLst[i] = False 

    self.IdleLock.set() #Makes sure no process is in Idle and all are ready to be terminated 

Answer 2

我会使用一个池。一个池有最大数量的线程它在同一时间使用，但你可以申请inf数量的作业。他们留在等待列表中，直到线程可用。我不认为你可以改变池中当前进程的数量。