Python的生产者消费者模式

gbless

#!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()