Python多线程学习(三、生产者与消费者)

 生产者与消费者问题是典型的同步问题。这里简单介绍两种不同的实现方法。

1，  条件变量

view plaincopy to clipboardprint?
import threading  
 
import time  
 
class Producer(threading.Thread):  
 
    def __init__(self, t_name):  
 
        threading.Thread.__init__(self, name=t_name)  
 
   
 
    def run(self):  
 
        global x  
 
        con.acquire()  
 
        if x > 0:  
 
            con.wait()  
 
        else:  
 
            for i in range(5):  
 
                x=x+1 
 
                print "producing..." + str(x)  
 
            con.notify()  
 
        print x  
 
        con.release()  
 
   
 
class Consumer(threading.Thread):  
 
    def __init__(self, t_name):  
 
        threading.Thread.__init__(self, name=t_name)  
 
    def run(self):  
 
        global x  
 
        con.acquire()  
 
        if x == 0:  
 
            print 'consumer wait1' 
 
            con.wait()  
 
        else:  
 
            for i in range(5):  
 
                x=x-1 
 
                print "consuming..." + str(x)  
 
            con.notify()  
 
        print x  
 
        con.release()  
 
   
 
con = threading.Condition()  
 
x=0 
 
print 'start consumer' 
 
c=Consumer('consumer')  
 
print 'start producer' 
 
p=Producer('producer')  
 
   
 
p.start()  
 
c.start()  
 
p.join()  
 
c.join()  
 
print x 
import threading

import time

class Producer(threading.Thread):

    def __init__(self, t_name):

        threading.Thread.__init__(self, name=t_name)

 

    def run(self):

        global x

        con.acquire()

        if x > 0:

            con.wait()

        else:

            for i in range(5):

                x=x+1

                print "producing..." + str(x)

            con.notify()

        print x

        con.release()

 

class Consumer(threading.Thread):

    def __init__(self, t_name):

        threading.Thread.__init__(self, name=t_name)

    def run(self):

        global x

        con.acquire()

        if x == 0:

            print 'consumer wait1'

            con.wait()

        else:

            for i in range(5):

                x=x-1

                print "consuming..." + str(x)

            con.notify()

        print x

        con.release()

 

con = threading.Condition()

x=0

print 'start consumer'

c=Consumer('consumer')

print 'start producer'

p=Producer('producer')

 

p.start()

c.start()

p.join()

c.join()

print x

 

    上面的例子中，在初始状态下，Consumer处于wait状态，Producer连续生产（对x执行增1操作）5次后，notify正在等待的Consumer。Consumer被唤醒开始消费（对x执行减1操作）

2，  同步队列

Python中的Queue对象也提供了对线程同步的支持。使用Queue对象可以实现多个生产者和多个消费者形成的FIFO的队列。

生产者将数据依次存入队列，消费者依次从队列中取出数据。

view plaincopy to clipboardprint?
# producer_consumer_queue  
 
from Queue import Queue  
 
import random  
 
import threading  
 
import time  
 
   
 
#Producer thread  
 
class Producer(threading.Thread):  
 
    def __init__(self, t_name, queue):  
 
        threading.Thread.__init__(self, name=t_name)  
 
        self.data=queue  
 
    def run(self):  
 
        for i in range(5):  
 
            print "%s: %s is producing %d to the queue!/n" %(time.ctime(), self.getName(), i)  
 
            self.data.put(i)  
 
            time.sleep(random.randrange(10)/5)  
 
        print "%s: %s finished!" %(time.ctime(), self.getName())  
 
   
 
#Consumer thread  
 
class Consumer(threading.Thread):  
 
    def __init__(self, t_name, queue):  
 
        threading.Thread.__init__(self, name=t_name)  
 
        self.data=queue  
 
    def run(self):  
 
        for i in range(5):  
 
            val = self.data.get()  
 
            print "%s: %s is consuming. %d in the queue is consumed!/n" %(time.ctime(), self.getName(), val)  
 
            time.sleep(random.randrange(10))  
 
        print "%s: %s finished!" %(time.ctime(), self.getName())  
 
   
 
#Main thread  
 
def main():  
 
    queue = Queue()  
 
    producer = Producer('Pro.', queue)  
 
    consumer = Consumer('Con.', queue)  
 
    producer.start()  
 
    consumer.start()  
 
    producer.join()  
 
    consumer.join()  
 
    print 'All threads terminate!' 
 
   
 
if __name__ == '__main__':  
 
    main() 
# producer_consumer_queue

from Queue import Queue

import random

import threading

import time

 

#Producer thread

class Producer(threading.Thread):

    def __init__(self, t_name, queue):

        threading.Thread.__init__(self, name=t_name)

        self.data=queue

    def run(self):

        for i in range(5):

            print "%s: %s is producing %d to the queue!/n" %(time.ctime(), self.getName(), i)

            self.data.put(i)

            time.sleep(random.randrange(10)/5)

        print "%s: %s finished!" %(time.ctime(), self.getName())

 

#Consumer thread

class Consumer(threading.Thread):

    def __init__(self, t_name, queue):

        threading.Thread.__init__(self, name=t_name)

        self.data=queue

    def run(self):

        for i in range(5):

            val = self.data.get()

            print "%s: %s is consuming. %d in the queue is consumed!/n" %(time.ctime(), self.getName(), val)

            time.sleep(random.randrange(10))

        print "%s: %s finished!" %(time.ctime(), self.getName())

 

#Main thread

def main():

    queue = Queue()

    producer = Producer('Pro.', queue)

    consumer = Consumer('Con.', queue)

    producer.start()

    consumer.start()

    producer.join()

    consumer.join()

    print 'All threads terminate!'

 

if __name__ == '__main__':

    main()

 

在上面的例子中，Producer在随机的时间内生产一个“产品”，放入队列中。Consumer发现队列中有了“产品”，就去消费它。本例中，由于Producer生产的速度快于Consumer消费的速度，所以往往Producer生产好几个“产品”后，Consumer才消费一个产品。

Queue模块实现了一个支持多producer和多consumer的FIFO队列。当共享信息需要安全的在多线程之间交换时，Queue非常有用。Queue的默认长度是无限的，但是可以设置其构造函数的maxsize参数来设定其长度。Queue的put方法在队尾插入，该方法的原型是：

put( item[, block[, timeout]])

如果可选参数block为true并且timeout为None（缺省值），线程被block，直到队列空出一个数据单元。如果timeout大于0，在timeout的时间内，仍然没有可用的数据单元，Full exception被抛出。反之，如果block参数为false（忽略timeout参数），item被立即加入到空闲数据单元中，如果没有空闲数据单元，Full exception被抛出。

Queue的get方法是从队首取数据，其参数和put方法一样。如果block参数为true且timeout为None（缺省值），线程被block，直到队列中有数据。如果timeout大于0，在timeout时间内，仍然没有可取数据，Empty exception被抛出。反之，如果block参数为false（忽略timeout参数），队列中的数据被立即取出。如果此时没有可取数据，Empty exception也会被抛出。

 

本文来自CSDN博客，转载请标明出处：http://blog.csdn.net/lazy_tiger/archive/2009/02/15/3893415.aspx