queue - python多线程队列处理任务的问题

问题描述

问 题

想用Queue和threading模块实现一个线程池，线程函数从任务队列取出任务，进行处理，处理结果可能会产生新的任务，根据条件放到任务队列。

class WorkManager(object):
    def __init__(self, arg, thread_num=2):
        self.work_queue = Queue.Queue()     # 队列对象
        self.threads = []
        self.__init_work_queue(arg)
        self.__init_thread_pool(thread_num)
    """
    初始化线程
    """
    def __init_thread_pool(self, thread_num):
        for i in range(thread_num):
            self.threads.append(Work(self.work_queue))
    """
    初始化工作队列
    """
    def __init_work_queue(self, url):
        self.work_queue.put(do_job, arg)


class Work(threading.Thread):
    def __init__(self, work_queue):
        threading.Thread.__init__(self)
        self.work_queue = work_queue
        self.start()

    def run(self):
        while True:
            try:
                do, arg = self.work_queue.get(block=False)  
                do(arg)
                self.work_queue.task_done()  # 通知系统任务完成
            except Exception, e:
                print self.getName(), "end", str(e)
                break


# 具体要做的任务
def do_job(args):
   # todo 

上述代码定义了任务队列的WorkManager类和产生线程的Work类。
处理任务的do_job函数由于在两个类中都要调用，所以只能写在类外。

所以我的问题是，do_job函数产生了新的任务时，如何能获取到work_queue这个Queue.Queue()对象，从而把任务添加到队列里。

解决方案

import time
import Queue
import threading
import traceback


class Worker(threading.Thread):
    def __init__(self, work_manager):
        threading.Thread.__init__(self)
        self.work_queue = work_manager.work_queue
        self.done_queue = work_manager.done_queue
        self.start()

    def run(self):
        while True:
            try:
                do, arg, kwargs = self.work_queue.get(block=True)
                result = do(*arg, **kwargs)
                self.done_queue.put((do, arg, result))
            except Exception, e:
                print traceback.format_exc()
                print self.getName(), "end", str(e)
                break


class WorkManager(object):
    def __init__(self, thread_num=2):
        self.work_queue = Queue.Queue()  # 队列对象
        self.done_queue = Queue.Queue()  # 执行完成的任务
        self.threads = []
        self._init_thread_pool(thread_num)

    def _init_thread_pool(self, thread_num):
        """初始化线程"""
        for i in range(thread_num):
            worker = Worker(self)
            self.threads.append(worker)

    def add_job(self, job, arg=[], kwargs={}):
        """初始化工作队列"""
        self.work_queue.put((job, arg, kwargs))


work_manager = WorkManager()
work_manager.add_job(sum, [[1, 2, 3]])
work_manager.add_job(max, [[2, 3, 4]])
work_manager.add_job(len, ['Hello World'])
time.sleep(1)
while not work_manager.done_queue.empty():
    print work_manager.done_queue.get(False)
    

1. 不需要do_job方法，通过work_manager.add_job来发布任务给线程池

2. 线程中Queue.get需要block

3. task_done方法是配合Queue.join使用的，这里应该不需要

4. 任务执行成功后把执行结果放在done_queue中

5. 方法的注释应该写在函数名的下面。

这篇关于queue - python多线程队列处理任务的问题的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！