python 生产者消费者模式使用condition

python_consume_produce_thread.py

#!/usr/bin/env python

import time
import thread
from threading import Thread, Condition, Event, Lock, currentThread
from Queue import Queue

class ItemQ(object):

    def __init__(self):
        self.count = 0

    def produce(self, num=1):
        self.count += num

    def consume(self):
        if self.count:
            self.count -= 1

    def isEmpty(self):
        return not self.count


class Producer(Thread):

    def __init__(self, condition, itemq, sleeptime=1):
        super(Producer, self).__init__()
        self.cond = condition
        self.itemq = itemq
        self.sleeptime = sleeptime

    def run(self):
        cond = self.cond
        itemq = self.itemq

        while 1:
            cond.acquire()
            print currentThread()
            itemq.produce()
            cond.notifyAll()
            cond.release()
            time.sleep(self.sleeptime) #switch from currentThread to another thread


class Consumer(Thread):

    def __init__(self, condition, itemq, sleeptime=2):
        super(Consumer, self).__init__()
        self.cond = condition
        self.itemq = itemq
        self.sleeptime = sleeptime

    def run(self):
        cond = self.cond
        itemq = self.itemq

        while 1:
            time.sleep(self.sleeptime) #because has two consume thread, sleep to switch another thread
            cond.acquire()
            while itemq.isEmpty():
                print 'cond wait '
                cond.wait()
                print 'cond wait cancel'

            itemq.consume()
            print currentThread()
            cond.release()


if __name__=="__main__":

    q = ItemQ()
    cond = Condition()
    
    pro = Producer(cond, q)
    cons1 = Consumer(cond, q)
    cons2 = Consumer(cond, q)

    pro.start()
    cons1.start()
    cons2.start()
    while 1:
        pass

#-*- coding:utf-8 -*-
#!/usr/bin/env python

import time
import thread
from threading import Thread, Condition, currentThread
import random

class Produce(Thread):

    def __init__(self, condition, items):
        super(Produce, self).__init__()
        self.condition = condition
        self.items = items

    def run(self):
        while 1:
            self.condition.acquire()
            num = random.randint(0, 100)
            print 'Produce', num
            self.items.append(num)
            self.condition.notifyAll() #notifyAll don't release lock
            self.condition.release()
            time.sleep(2) #switch to consume


class Consume(Thread):

    def __init__(self, condition, items):
        super(Consume, self).__init__()
        self.condition = condition
        self.items = items

    def run(self):
        while 1:
            self.condition.acquire()
            if not self.items:
                print 'Consume start wait'
                self.condition.wait() #释放锁并且阻塞当前线程，当被唤醒时又重新获得锁，由于使用时Rlock 重用锁，所以才会调用 release
                print 'Consume end wait'

            print 'Consume', self.items.pop()
            print 'consume condition release'
            self.condition.release()

if __name__ == '__main__':
    c1 = Condition()
    items = []
    p = Produce(c1, items)
    c = Consume(c1, items)
    p.start()
    c.start()
    p.join()
    c.join()
    while 1:
        pass