Python 스레드 풀 threadpool 구현-파이썬 튜토리얼-php.cn

이 글에서는 주로 Python 스레드 풀 threadpool의 구현을 자세히 소개하며, 관심 있는 친구들은 참고할 수 있습니다.

이 글은 참고용으로 스레드 풀의 모든 작업을 공유합니다.

먼저 제가 사용하는 명사를 소개하겠습니다.

Worker: 스레드 풀을 생성할 때 지정된 스레드 수에 따라 작업자 스레드를 생성하고 작업 대기열에서 작업을 가져오기를 기다립니다.

Task(요청): 즉, 작업자 스레드에 의해 처리되는 작업은 수천 개가 될 수 있지만 작업자 스레드는 소수에 불과합니다. 작업은 makeRequests를 통해 생성됩니다.

작업 대기열(request_queue): 작업을 저장하기 위한 대기열이며 대기열을 사용하여 구현됩니다. 작업자 스레드는 작업 대기열에서 작업을 가져와 처리합니다.

작업 처리 함수(호출 가능): 작업자 스레드는 작업을 가져온 후 작업 처리 함수(request.callable_)를 호출하여 작업을 구체적으로 처리하고 처리 결과를 반환합니다.

작업 결과 대기열(result_queue): 작업 처리가 완료된 후 반환된 처리 결과는 작업 결과 대기열에 저장됩니다(예외 포함).

작업 예외 처리 함수 또는 콜백(exc_callback): 가져오기 결과적으로, 예외가 설정되면 예외를 처리하기 위해 예외 콜백을 호출해야 합니다.

작업 결과 콜백(콜백): 작업 결과 대기열에서 결과를 가져와서 추가로 처리합니다. 이전 섹션에서는 스레드 풀 스레드풀의 설치 및 사용을 소개했으며, 이번 섹션에서는 주로 스레드 풀 작업의 주요 프로세스를 소개합니다.

(1) 스레드 풀 생성

(2) 작업자 스레드 시작

(3) 스레드 풀 생성 작업
(4) 작업을 스레드 풀로 푸시
(5) 스레드 처리 작업
(6) 작업 종료 처리
(7) 작업자 스레드 종료

다음은 스레드 풀의 정의입니다.

class ThreadPool: 
  """A thread pool, distributing work requests and collecting results. 
 
  See the module docstring for more information. 
 
  """ 
  def __init__(self, num_workers, q_size=0, resq_size=0, poll_timeout=5): 
    pass 
  def createWorkers(self, num_workers, poll_timeout=5): 
    pass 
  def dismissWorkers(self, num_workers, do_join=False): 
    pass 
  def joinAllDismissedWorkers(self): 
    pass 
  def putRequest(self, request, block=True, timeout=None): 
    pass 
  def poll(self, block=False): 
    pass 
  def wait(self): 
    pass

로그인 후 복사

1. 스레드 풀 생성(ThreadPool(args ))

task_pool=threadpool.ThreadPool(num_works)

task_pool=threadpool.ThreadPool(num_works) 
  def __init__(self, num_workers, q_size=0, resq_size=0, poll_timeout=5): 
    """Set up the thread pool and start num_workers worker threads. 
 
    ``num_workers`` is the number of worker threads to start initially. 
 
    If ``q_size > 0`` the size of the work *request queue* is limited and 
    the thread pool blocks when the queue is full and it tries to put 
    more work requests in it (see ``putRequest`` method), unless you also 
    use a positive ``timeout`` value for ``putRequest``. 
 
    If ``resq_size > 0`` the size of the *results queue* is limited and the 
    worker threads will block when the queue is full and they try to put 
    new results in it. 
 
    .. warning: 
      If you set both ``q_size`` and ``resq_size`` to ``!= 0`` there is 
      the possibilty of a deadlock, when the results queue is not pulled 
      regularly and too many jobs are put in the work requests queue. 
      To prevent this, always set ``timeout > 0`` when calling 
      ``ThreadPool.putRequest()`` and catch ``Queue.Full`` exceptions. 
 
    """ 
    self._requests_queue = Queue.Queue(q_size)#任务队列，通过threadpool.makeReuests(args)创建的任务都会放到此队列中 
    self._results_queue = Queue.Queue(resq_size)#字典，任务对应的任务执行结果</span> 
    self.workers = []#工作线程list，通过self.createWorkers()函数内创建的工作线程会放到此工作线程list中 
    self.dismissedWorkers = []#被设置线程事件并且没有被join的工作线程 
    self.workRequests = {}#字典，记录任务被分配到哪个工作线程中</span> 
    self.createWorkers(num_workers, poll_timeout)

로그인 후 복사

그중 초기화 매개변수는 다음과 같습니다.

num_works: 스레드 수 스레드 풀

q_size: 작업 대기열의 길이 제한, 대기열의 길이가 제한된 경우 작업을 추가하기 위해 putRequest()가 호출될 때 제한 길이에 도달한 후 putRequest는 작업 추가를 계속 시도합니다. putRequest()에 시간 초과 또는 차단이 설정되어 있습니다.

esq_size: 작업 결과 대기열의 길이
pool_timeout: 작업자 스레드가 요청 대기열에서 요청을 읽을 수 없으면 pool_timeout을 차단합니다.

그 중 멤버 변수:

self._requests_queue: 작업 대기열, threadpool.makeReuests(args) 생성된 작업은 이 대기열에 배치됩니다.

self._results_queue: 사전, 해당 작업 실행 task

self.workers: 작업자 스레드 목록, self.createWorkers() 함수를 통해 생성된 작업자 스레드는 이 작업자 스레드 목록에 배치됩니다. Medium;
self.dismisssedWorkers: 스레드 이벤트가 설정되고 작업자 스레드가 조인되지 않습니다. self.workRequests: 스레드 풀에 푸시된 작업을 기록하는 사전, 구조는 requestID:request입니다. requestID는 나중에 소개할 작업의 고유 식별자입니다.

2. 작업자 스레드 시작(self.createWorks(args))

함수 정의:

def createWorkers(self, num_workers, poll_timeout=5): 
   """Add num_workers worker threads to the pool. 
 
   ``poll_timout`` sets the interval in seconds (int or float) for how 
   ofte threads should check whether they are dismissed, while waiting for 
   requests. 
 
   """ 
   for i in range(num_workers): 
     self.workers.append(WorkerThread(self._requests_queue, 
       self._results_queue, poll_timeout=poll_timeout))

로그인 후 복사

WorkerThread()는 Python의 내장 스레드 클래스인 스레드에서 상속됩니다. WorkerThread 객체를 생성합니다. 이를 self.workers 대기열에 넣습니다. WorkerThread 클래스의 정의를 살펴보겠습니다.

self.__init__(args)에서 볼 수 있습니다:

class WorkerThread(threading.Thread): 
  """Background thread connected to the requests/results queues. 
 
  A worker thread sits in the background and picks up work requests from 
  one queue and puts the results in another until it is dismissed. 
 
  """ 
 
  def __init__(self, requests_queue, results_queue, poll_timeout=5, **kwds): 
    """Set up thread in daemonic mode and start it immediatedly. 
 
    ``requests_queue`` and ``results_queue`` are instances of 
    ``Queue.Queue`` passed by the ``ThreadPool`` class when it creates a 
    new worker thread. 
 
    """ 
    threading.Thread.__init__(self, **kwds) 
    self.setDaemon(1)# 
    self._requests_queue = requests_queue#任务队列 
    self._results_queue = results_queue#任务结果队列 
    self._poll_timeout = poll_timeout#run函数中从任务队列中get任务时的超时时间，如果超时则继续while(true)； 
    self._dismissed = threading.Event()#线程事件，如果set线程事件则run会执行break,直接退出工作线程； 
    self.start() 
 
  def run(self): 
    """Repeatedly process the job queue until told to exit.""" 
    while True: 
      if self._dismissed.isSet():#如果设置了self._dismissed则退出工作线程 
 
        # we are dismissed, break out of loop 
        break 
      # get next work request. If we don&#39;t get a new request from the 
      # queue after self._poll_timout seconds, we jump to the start of 
      # the while loop again, to give the thread a chance to exit. 
      try: 
        request = self._requests_queue.get(True, self._poll_timeout) 
      except Queue.Empty:#尝从任务 队列self._requests_queue 中get任务，如果队列为空，则continue 
        continue 
      else: 
        if self._dismissed.isSet():#检测此工作线程事件是否被set,如果被设置，意味着要结束此工作线程，那么就需要将取到的任务返回到任务队列中，并且退出线程 
          # we are dismissed, put back request in queue and exit loop 
          self._requests_queue.put(request) 
          break 
        try:<span style="color:#如果线程事件没有被设置，那么执行任务处理函数request.callable，并将返回的result，压入到任务结果队列中 
          result = request.callable(*request.args, **request.kwds) 
          self._results_queue.put((request, result)) 
        except: 
          request.exception = True 
          self._results_queue.put((request, sys.exc_info()))#如果任务处理函数出现异常，则将异常压入到队列中 
 
  def dismiss(self):</span> 
    """Sets a flag to tell the thread to exit when done with current job. 
    """ 
    self._dismissed.set()

로그인 후 복사

초기화 변수:

self._request_queue: 작업 대기열; _resutls_queuqe,: 작업 결과 Queue;
self._pool_timeout: 실행 함수에서 작업 대기열에서 작업을 가져올 때의 시간 초과입니다. while(true): 스레드 이벤트인 경우.

마지막으로 self.start()를 호출하여 스레드를 시작합니다.

위에서 실행 함수의 실행 단계는 다음과 같습니다.

(1) self._dismissed가 설정되어 있으면 작업 스레드를 종료하고, 그렇지 않으면 2단계를 실행합니다.

(2) 작업 대기열 self._requests_queue에서 작업을 가져옵니다. 대기열이 비어 있으면 계속 실행합니다. 다음 while 루프를 실행하고 그렇지 않으면 3단계를 실행합니다.
(3) 이 작업자 스레드 이벤트가 설정되었는지 확인합니다. 설정되었다는 것이 이 작업자 스레드를 종료한다는 것을 의미하는 경우 가져온 작업을 작업 대기열로 반환하고 스레드를 종료해야 합니다. 스레드 이벤트가 설정되지 않은 경우 작업 처리 함수 request.callable이 실행되고 반환된 결과가 작업 결과 큐에 푸시됩니다. 작업 처리 함수에서 예외가 발생하면 해당 예외가 큐에 푸시됩니다. 마지막으로 4

(4)단계로 점프하여 루프를 계속하고 1

으로 돌아갑니다.

到此工作线程创建完毕，根据设置的线程池线程数量，创建工作线程，工作线程从任务队列中get任务，进行任务处理，并将任务处理结果压入到任务结果队列中。

3、任务的创建（makeRequests）

任务的创建函数为threadpool.makeRequests(callable_,args_list,callback=None):

# utility functions 
def makeRequests(callable_, args_list, callback=None, 
    exc_callback=_handle_thread_exception): 
  """Create several work requests for same callable with different arguments. 
 
  Convenience function for creating several work requests for the same 
  callable where each invocation of the callable receives different values 
  for its arguments. 
 
  ``args_list`` contains the parameters for each invocation of callable. 
  Each item in ``args_list`` should be either a 2-item tuple of the list of 
  positional arguments and a dictionary of keyword arguments or a single, 
  non-tuple argument. 
 
  See docstring for ``WorkRequest`` for info on ``callback`` and 
  ``exc_callback``. 
 
  """ 
  requests = [] 
  for item in args_list: 
    if isinstance(item, tuple): 
      requests.append( 
        WorkRequest(callable_, item[0], item[1], callback=callback, 
          exc_callback=exc_callback) 
      ) 
    else: 
      requests.append( 
        WorkRequest(callable_, [item], None, callback=callback, 
          exc_callback=exc_callback) 
      ) 
  return requests

로그인 후 복사

其中创建任务的函数参数具体意义为下：

callable_：注册的任务处理函数，当任务被放到任务队列后，工作线程中获取到该任务的线程，会执行此 callable_

args_list：首先args_list是列表，列表元素类型为元组，元组中有两个元素item[0]，item[1]，item[0]为位置参数，item[1]为字典类型关键字参数。列表中元组的个数，代表启动的任务个数，在使用的时候一般都为单个元组，即一个makerequest()创建一个任务。

callback：回调函数，在poll函数中调用（后面讲解此函数），callable_调用结束后，会就任务结果放入到任务结果队列中（self._resutls_queue），在poll函数中，当从self._resutls_queue队列中get某个结果后，会执行此callback(request,result),其中result是request任务返回的结果。

exc_callback：异常回调函数，在poll函数中，如果某个request对应有执行异常，那么会调用此异常回调。

创建完成任务后，返回创建的任务。

外层记录此任务，放入到任务列表中。

上面是创建任务的函数，下面讲解任务对象的结构：

class WorkRequest: 
  """A request to execute a callable for putting in the request queue later. 
 
  See the module function ``makeRequests`` for the common case 
  where you want to build several ``WorkRequest`` objects for the same 
  callable but with different arguments for each call. 
 
  """ 
 
  def __init__(self, callable_, args=None, kwds=None, requestID=None, 
      callback=None, exc_callback=_handle_thread_exception): 
    """Create a work request for a callable and attach callbacks. 
 
    A work request consists of the a callable to be executed by a 
    worker thread, a list of positional arguments, a dictionary 
    of keyword arguments. 
 
    A ``callback`` function can be specified, that is called when the 
    results of the request are picked up from the result queue. It must 
    accept two anonymous arguments, the ``WorkRequest`` object and the 
    results of the callable, in that order. If you want to pass additional 
    information to the callback, just stick it on the request object. 
 
    You can also give custom callback for when an exception occurs with 
    the ``exc_callback`` keyword parameter. It should also accept two 
    anonymous arguments, the ``WorkRequest`` and a tuple with the exception 
    details as returned by ``sys.exc_info()``. The default implementation 
    of this callback just prints the exception info via 
    ``traceback.print_exception``. If you want no exception handler 
    callback, just pass in ``None``. 
 
    ``requestID``, if given, must be hashable since it is used by 
    ``ThreadPool`` object to store the results of that work request in a 
    dictionary. It defaults to the return value of ``id(self)``. 
 
    """ 
    if requestID is None: 
      self.requestID = id(self) 
    else: 
      try: 
        self.requestID = hash(requestID) 
      except TypeError: 
        raise TypeError("requestID must be hashable.") 
    self.exception = False 
    self.callback = callback 
    self.exc_callback = exc_callback 
    self.callable = callable_ 
    self.args = args or [] 
    self.kwds = kwds or {} 
 
  def __str__(self): 
    return "<WorkRequest id=%s args=%r kwargs=%r exception=%s>" % \ 
      (self.requestID, self.args, self.kwds, self.exception)

로그인 후 복사

上面self.callback 以及self.exc_callback，和self.callable_ ,args,dwds都已经讲解，就不在啰嗦了。
其中有一个任务的全局唯一标识，即self.requestID,通过获取自身内存首地址作为自己的唯一标识id(self)
self.exception 初始化为False,如果执行self.callable()过程中出现异常，那么此变量会标设置为True。

至此，任务创建完毕，调用makeRequests（）的上层记录有任务列表request_list.

4、任务的推送到线程池（putRequest）

上面小节中介绍了任务的创建，任务的个数可以成千上百，但是处理任务的线程数量只有我们在创建线程池的时候制定的线程数量来处理，指定的线程数量往往比任务的数量小得多，因此，每个线程必须处理多个任务。

本节介绍如何将创建的任务推送的线程池中，以让线程池由阻塞状态，获取任务，然后去处理任务。
任务的推送使用ThreadPool线程池类中的putRequest（self,request,block,timeout）来创建：

def putRequest(self, request, block=True, timeout=None): 
  """Put work request into work queue and save its id for later.""" 
  assert isinstance(request, WorkRequest) 
  # don&#39;t reuse old work requests 
  assert not getattr(request, &#39;exception&#39;, None) 
  self._requests_queue.put(request, block, timeout) 
  self.workRequests[request.requestID] = request

로그인 후 복사

函数的主要作用就是将request任务，也就是上一小节中创建的任务，put到线程池的任务队列中（self._request_queue）。然后记录已经推送到线程池的任务，通过线程池的self.workReuests 字典来存储，结构为request.requestID:request。

至此，任务创建完成，并且已经将任务推送到线程池中。

5、线程处理任务

通过上一小节，任务已经推送到了线程中。在任务没有被推送到线程池中时，线程池中的线程都处在处在阻塞状态中，即在线程的self.run（）函数中，一直处于一下状态：

try: 
  request = self._requests_queue.get(True, self._poll_timeout) 
except Queue.Empty:#尝从任务 队列self._requests_queue 中get任务，如果队列为空，则continue 
  continue

로그인 후 복사

现在任务已经推送到线程池中，那么get任务将会正常返回，会执行下面的步骤：

def run(self): 
    """Repeatedly process the job queue until told to exit.""" 
    while True: 
      if self._dismissed.isSet():#如果设置了self._dismissed则退出工作线程 
 
        # we are dismissed, break out of loop 
        break 
      # get next work request. If we don&#39;t get a new request from the 
      # queue after self._poll_timout seconds, we jump to the start of 
      # the while loop again, to give the thread a chance to exit. 
      try: 
        request = self._requests_queue.get(True, self._poll_timeout) 
      except Queue.Empty:#尝从任务 队列self._requests_queue 中get任务，如果队列为空，则continue 
        continue 
      else: 
        if self._dismissed.isSet():#检测此工作线程事件是否被set,如果被设置，意味着要结束此工作线程，那么就需要将取到的任务返回到任务队列中，并且退出线程 
          # we are dismissed, put back request in queue and exit loop 
          self._requests_queue.put(request) 
          break 
        try:#如果线程事件没有被设置，那么执行任务处理函数request.callable，并将返回的result，压入到任务结果队列中 
          result = request.callable(*request.args, **request.kwds) 
          self._results_queue.put((request, result)) 
        except: 
          request.exception = True 
          self._results_queue.put((request, sys.exc_info()))#如果任务处理函数出现异常，则将异常压入到队列中

로그인 후 복사

获取任务--->调用任务的处理函数callable()处理任务--->将任务request以及任务返回的结果压入到self.results_queue队列中---->如果任务处理函数异常，那么将任务异常标识设置为True,并将任务request以及任务异常压入到self.results_queue队列中---->再次返回获取任务

如果，在while循环过程中，外部设置了线程事件，即self._dismissed.isSet为True,那么意味着此线程将会结束处理任务，那么会将get到的任务返回的任务队列中，并且退出线程。

6、任务结束处理

上面小节中，介绍了线程池不断的get任务，并且不断的处理任务。那么每个任务结束之后我们该怎么处理呢，线程池提供了wait（）以及poll（）函数。

当我们把任务提交个线程池之后，我们会调用wait（）来等待任务处理结束，结束后wait()将会返回，返回后我们可以进行下一步操作，例如重新创建任务，将任务继续推送到线程池中，或者结束线程池。结束线程池会在下一小节介绍，这一小节主要介绍wait()和poll（）操作。

先来看看wait()操作：

def wait(self): 
  """Wait for results, blocking until all have arrived.""" 
  while 1: 
    try: 
      self.poll(True) 
    except NoResultsPending: 
      break

로그인 후 복사

等待任务处理结束，在所有任务处理完成之前一直处于block阶段，如果self.poll()返回异常NoResultsPending异常，然后wait返回，任务处理结束。
下面看看poll函数：

def poll(self, block=False): 
  """Process any new results in the queue.""" 
  while True: 
    # still results pending? 
    if not self.workRequests: 
      raise NoResultsPending 
    # are there still workers to process remaining requests? 
    elif block and not self.workers: 
      raise NoWorkersAvailable 
    try: 
      # get back next results 
      request, result = self._results_queue.get(block=block) 
      # has an exception occured? 
      if request.exception and request.exc_callback: 
        request.exc_callback(request, result) 
      # hand results to callback, if any 
      if request.callback and not \ 
          (request.exception and request.exc_callback): 
        request.callback(request, result) 
      del self.workRequests[request.requestID] 
    except Queue.Empty: 
      break

로그인 후 복사

（1）首先，检测任务字典（{request.requestID:request}）是否为空，如果为空则抛出异常NoResultPending结束，否则到第2步；

（2）检测工作线程是否为空（如果某个线程的线程事件被设置，那么工作线程退出，并从self.workers中pop出），如果为空则抛出NoWorkerAvailable异常结束，否则进入第3步；

（3）从任务结果队列中get任务结果，如果抛出队列为空，那么break,返回，否则进入第4步；

（4）如果任务处理过程中出现异常，即设置了request.exception，并且设置了异常处理回调即request.exc_callback则执行异常回调，再回调中处理异常，返回后将任务从任务列表self.workRequests中移除，继续get任务，返回第1步。否则进入第5步；

（5）如果设置了任务结果回调即request.callback不为空，则执行任务结果回调即request.callbacl(request,result),并
将任务从任务列表self.workRequests中移除，继续get任务，返回第1步。

（6）重复进行上面的步骤直到抛出异常，或者任务队列为空，则poll返会；

至此抛出NoResultPending wait操作接受此异常后，至此wait()返回。

7、工作线程的退出

threadpool提供的工作线程退出的的操作有dismissWorkers()和joinAllDismissedWorker()操作：

def dismissWorkers(self, num_workers, do_join=False): 
  """Tell num_workers worker threads to quit after their current task.""" 
  dismiss_list = [] 
  for i in range(min(num_workers, len(self.workers))): 
    worker = self.workers.pop() 
    worker.dismiss() 
    dismiss_list.append(worker) 
 
  if do_join: 
    for worker in dismiss_list: 
      worker.join() 
  else: 
    self.dismissedWorkers.extend(dismiss_list) 
 
def joinAllDismissedWorkers(self): 
  """Perform Thread.join() on all worker threads that have been dismissed. 
  """ 
  for worker in self.dismissedWorkers: 
    worker.join() 
  self.dismissedWorkers = []

로그인 후 복사

从dismissWorkers可看出，主要工作是从self.workers 工作线程中pop出指定的线程数量，并且设置此线程的线程事件，设置线程事件后，此线程self.run()函数，则会检测到此设置，并结束线程。

如果设置了在do_join，即设置了在此函数中join退出的线程，那么对退出的线程执行join操作。否则将pop出的线程放入到self.dismissedWorkers中，以等待joinAllDismissedWorkers操作去处理join线程。

8、总结

到此为止，threadpool线程池中所有的操作介绍完毕，其实现也做了具体的介绍。从上面可看出，线程池并没有那么复杂，只有几个简单的操作，主要是了解整个处理流程即可。

希望大家多多提出建议和意见。