Python的生产者消费者模式

#!python
# -*- coding: utf-8 -*-
"""
Description: 生产者/消费者模式的模块
File: thread_worker.py
Authors: kanpiaoxue
Date: 2015年11月12日 下午5:51:08
"""

from Queue import Queue
import logging
import threading
import time

import logsetting


class TaskProtocol(object):
    """
            执行任务协议类
    """

    def execute(self):
        """
                    执行任务
        """

        pass


class ProduceTasksProtocol(object):
    """
            产出运行任务(TaskProtocol)的协议类
    """

    def getTasks(self):
        """
                    产出运行的任务(TaskProtocol)列表
        @return: 任务列表
        """

        pass


class Producer(threading.Thread):
    """
        生产者消/费者模式：生产者类
    """

    def __init__(self, name, produceTasksProtocol, queue, sleepTime = 60):
        """
        @param name: 线程名称
        @param produceTasksProtocol: 生成 任务列表的协议类
        @param queue: 运行队列
        @param sleepTime: 生产者两次运行之间的间隔时间，单位：秒
        """

        threading.Thread.__init__(self, name = name)
        self.threadStop = False
        self.name = name
        self.produceTasksProtocol = produceTasksProtocol
        self.queue = queue
        self.sleepTime = int(sleepTime)

    def run(self):
        """
                    线程运行方法
        """

        logging.info('thread %s start to work.' % self.name)
        while not self.threadStop:
            tasks = self.produceTasksProtocol.getTasks()
            size = str(len(tasks))
            logging.info('produce %s tasks.' % size)
            for task in tasks:
                self.queue.put(task)
                logging.info('%s was put into queue.' % task)
            logging.info('%s will sleep %s.' % (self.name, self.sleepTime))
            time.sleep(self.sleepTime)

    def stop(self):
        """
        停止线程运行
        """

        self.threadStop = True

class Consumer(threading.Thread):
    """
        生产者消/费者模式：消费者类
    """

    def __init__(self, name, queue):
        """
        @param name: 线程名称
        @param queue: 运行队列
        """

        threading.Thread.__init__(self, name = name)
        self.threadStop = False
        self.name = name
        self.queue = queue

    def run(self):
        """
                    线程运行方法
        """

        logging.info('thread %s start to work.' % (self.name))
        while not self.threadStop:
            task = self.queue.get()
            logging.info('%s was take from queue and to work.' % task)
            start = time.clock()
            try:
                task.execute()
            except Exception as error:
                # 捕获异常，并记录错误日志。防止因为一个任务的失败造成线程退出工作
                logging.error('when execute task, Occur some exception: %s' % error)
            end = time.clock()
            logging.info('%s finish working. It consumes %s seconds.' % (task,str((end - start))))

    def stop(self):
        """
        停止线程运行
        """

        self.threadStop = True

if __name__ == '__main__':
    logsetting.init_log("./log/ctapi")
    class TaskProtocol_01(TaskProtocol):
        """
                执行任务协议类
        """
        def __init__(self, num):
            TaskProtocol.__init__(self)
            self.num = num

        def execute(self):
            """
                        执行任务
            """
            print self, 'start to work'

        def  __str__(self):
            return 'task-' + str(self.num)

    class ProduceTasksProtocol_01(ProduceTasksProtocol):
        """
                产出运行任务(TaskProtocol)的协议类
        """

        def __init__(self, count):
            ProduceTasksProtocol.__init__(self)
            self.count = count

        def getTasks(self):
            """
                        产出运行的任务(TaskProtocol)列表
            @return: 任务列表
            """
            return [TaskProtocol_01(i) for i in xrange(count)]

    count = 100
    produceTasksProtocol = ProduceTasksProtocol_01(count)
    queue = Queue()
    producer = Producer('producer', produceTasksProtocol, queue,sleepTime=5)
    producer.start()
    consumerCount = 5
    for x in range(consumerCount) :
        consumer = Consumer('consumer-' + str(x), queue)
        consumer.start()
    
    threading.current_thread().join()