devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

利用python实现生产者消费者的并发模型_1.使用python面向的对象写法,编写多线程生产消费者并发模型

作者：Monodyee | 2024-02-27 13:55:41

赞

踩

1.使用python面向的对象写法,编写多线程生产消费者并发模型

一、使用多线程实现生产者与消费者模型

1、Condition模型

可以认为Condition对象维护了一个锁（Lock/RLock)和一个waiting池。线程通过acquire获得Condition对象，当调用wait方法时，线程会释放Condition内部的锁并进入blocked状态，同时在waiting池中记录这个线程。当调用notify方法时，Condition对象会从waiting池中挑选一个线程，通知其调用acquire方法尝试取到锁，但是notify and notifyall本身是不会释放占有的Condition内部的锁，所以随后需要condition.release()来显示的释放锁。

Condition对象的构造函数可以接受一个Lock/RLock对象作为参数，如果没有指定，则Condition对象会在内部自行创建一个RLock。

  1. import threading
  2. import time,random
  3.  
  4. product=[];
  5. cond = threading.Condition()
  6.  
  7. class Producer(threading.Thread):
  8.     def __init__(self,speed):
  9.         threading.Thread.__init__(self)
  10.         self.speed = speed
  11.     def run(self):
  12.         while True:
  13.             for i in range(self.speed): 
  14.                 cond.acquire()
  15.                 while len(product)>=5:
  16.                     cond.wait();
  17.                 num=random.random();
  18.                 product.append(num)
  19.                 print("生产者"+str(self.speed)+"生产了："+str(num))
  20.                 cond.notifyAll()
  21.                 cond.release()
  22.             time.sleep(1)
  23.  
  24. class Consumer(threading.Thread):
  25.     def __init__(self):
  26.         threading.Thread.__init__(self)
  27.  
  28.     def run(self):
  29.         count=0
  30.         time_start=time.clock() 
  31.         while True:         
  32.             speed=random.randint(1,5);
  33.             for i in range(speed):
  34.                 cond.acquire()
  35.                 while len(product)<=0:
  36.                     cond.wait()
  37.                 num=product[0];
  38.                 del product[0]
  39.                 print("消费者，消费了"+str(num))
  40.                 count=count+1       
  41.                 if(count==20):                            
  42.                     time_end=time.clock()        
  43.                     print("消费 "+str(count)+"个商品所用时间: %f s" % (time_end - time_start))
  44.                 cond.notify()  
  45.                 cond.release()
  46.             time.sleep(1)
  47.  
  48.  
  49.  
  50. Producer(1).start();
  51. Producer(2).start();
  52. Consumer().start();

2、Queue模型

（1）创建一个 Queue.Queue() 的实例，来表示缓冲池。

（2）每次从使用生产者线程对队列中的数据进行填充，使用消费者线程取出队列中的数据。

  1. import threading
  2. import queue
  3. import time
  4. import random
  5.  
  6.  
  7. class Producer(threading.Thread):
  8.     def __init__(self,speed,queue):
  9.         threading.Thread.__init__(self)
  10.         self.speed = speed
  11.         self.queue = queue
  12.  
  13.     def run(self):
  14.         while True:
  15.             for i in range(self.speed):
  16.                 item = random.random()
  17.                 self.queue.put(item)
  18.                 print("生产者",self.speed,"生产了：",str(item))
  19.             time.sleep(1)
  20.             
  21. class Consumer(threading.Thread):
  22.     def __init__(self,queue):
  23.         threading.Thread.__init__(self)
  24.         self.queue = queue
  25.  
  26.     def run(self):
  27.         time_start=time.clock()       
  28.         count=0
  29.         while True:
  30.             speed=random.randint(1, 5)
  31.             for i in range(speed):
  32.                 item = self.queue.get()
  33.                 print("消费者","消费：",item)
  34.                 count=count+1       
  35.                 if(count==20):                            
  36.                     time_end=time.clock()        
  37.                     print("消费 "+str(count)+"个商品所用时间: %f s" % (time_end - time_start))
  38.             time.sleep(1)
  39.  
  40. q = queue.Queue(maxsize = 5)
  41.  
  42.  
  43. if __name__ == '__main__':
  44.     
  45.     Producer(1,q).start()
  46.     
  47.     Producer(2,q).start()
  48.     
  49.     Consumer(q).start()

3、信号量模型：

  1. import sys, time
  2. import random
  3. from threading import Thread, Semaphore
  4.  
  5. product = [] 
  6.  
  7. mutex = Semaphore(1)
  8. #也可以是mutex = RLock()
  9. full = Semaphore(0)
  10. empty = Semaphore(5)
  11.  
  12. class Producer(Thread):
  13.     def __init__(self, speed):
  14.         Thread.__init__(self);
  15.         self.speed=speed;
  16.  
  17.     def run(self):
  18.         while True:
  19.             for i in range(self.speed):
  20.                 ProductThing=''; 
  21.                 empty.acquire()
  22.                 mutex.acquire()
  23.                 num=random.random()                       
  24.                 ProductThing+=str(num)+' '                                        
  25.                 product.append(str(num))
  26.                 print ('%s: count=%d' % ("producer"+str(self.speed), len(product)))
  27.                 print ('         product things:'+ProductThing)  
  28.                 mutex.release()
  29.                 full.release()
  30.             time.sleep(1)
  31.  
  32.  
  33. class Consumer(Thread):
  34.     def __init__(self):
  35.         Thread.__init__(self);
  36.  
  37.     def run(self):
  38.         count=0
  39.         time_start=time.clock() 
  40.         while True:
  41.             speed=random.randint(1, 5)
  42.             for i in range(speed):
  43.                 consumeThing=""     
  44.                 full.acquire()
  45.                 mutex.acquire()
  46.                 consumeThing+=str(product[0])+' '
  47.                 del product[0]     
  48.                 print('%s: count=%d' % ("consumer", len(product)))
  49.                 print('         consume things:'+consumeThing)
  50.                 count=count+1       
  51.                 if(count==20):                            
  52.                     time_end=time.clock()        
  53.                     print("消费 "+str(count)+"个商品所用时间: %f s" % (time_end - time_start))                                                                                       
  54.                 mutex.release()
  55.                 empty.release()
  56.             time.sleep(1)
  57.  
  58.  
  59. if __name__ == '__main__':
  60.  
  61.     Producer(1).start()
  62.     
  63.     Producer(2).start()
  64.  
  65.     Consumer().start()

4、Event模型

threading.Event机制类似于一个线程向其它多个线程发号施令的模式，其它线程都会持有一个threading.Event的对象，这些线程都会等待这个事件的“发生”，如果此事件一直不发生，那么这些线程将会阻塞，直至事件的“发生”。生产者生产完商品会立即通知消费者去消费，消费者消费完商品后会立即通知生产者去生产，适用于产品池数目为一的情况。

  1. import threading 
  2. import random
  3. import time
  4.  
  5. def produce(speed,e_p1,e_p2,e_c,product):
  6.     while True:
  7.         for i in range(speed):
  8.             if(speed==1):
  9.                 e_p1.wait();
  10.             if(speed==2):
  11.                 e_p2.wait(); 
  12.             num=random.random();
  13.             product.append(num)
  14.             print("生产者"+str(speed)+",生产了"+str(num))
  15.             e_c.set();
  16.             if(speed==1):
  17.                 e_p1.clear();
  18.             if(speed==2):
  19.                 e_p2.clear(); 
  20.         time.sleep(1)
  21.  
  22. def consume(e_p1,e_p2,e_c,product):
  23.     count=0
  24.     time_start=time.clock() 
  25.     while True:
  26.         speed=random.randint(1,5);
  27.         for i in range(speed):
  28.             e_c.wait();
  29.             num=product[0];
  30.             del product[0]
  31.             print("消费者，消费了"+str(num))
  32.             count=count+1       
  33.             if(count==20):                            
  34.                 time_end=time.clock()        
  35.                 print("消费 "+str(count)+"个商品所用的时间: %f s" % (time_end - time_start))
  36.             r=random.randint(1,2)
  37.             if(r==1):
  38.                 e_p1.set()
  39.             if(r==2):
  40.                 e_p2.set() 
  41.             e_c.clear()
  42.         time.sleep(1)
  43.  
  44. if __name__ == '__main__':
  45.     
  46.     e_p1= threading.Event();
  47.     
  48.     e_p2= threading.Event();
  49.     
  50.     e_c= threading.Event();
  51.     
  52.     e_p1.set()
  53.     
  54.     product=[]
  55.     
  56.     p1=threading.Thread(target=produce, args=(1,e_p1,e_p2,e_c,product))
  57.     
  58.     p1.start()  
  59.  
  60.     p2=threading.Thread(target=produce, args=(2,e_p1,e_p2,e_c,product))
  61.     
  62.     p2.start()
  63.     
  64.     c1=threading.Thread(target=consume, args=(e_p1,e_p2,e_c,product))
  65.     
  66.     c1.start()

二、多进程实现生产者与消费者模型

1、  信号量和共享内存

  1. '''
  2. @author: jqy
  3. '''
  4.  
  5. from multiprocessing import Process
  6. import time
  7. import random
  8. import multiprocessing
  9.  
  10. def produce(speed,mutex,full,empty,product,pindex,cindex):
  11.     while True:
  12.         for i in range(speed):
  13.             ProductThing='';
  14.             empty.acquire()
  15.             mutex.acquire()
  16.             num=random.random()
  17.             ProductThing+=str(num)+' '
  18.             product[pindex.value]=num
  19.             pindex.value=(pindex.value+1)%len(product)
  20.             print ('%s: product things:%s' % ("producer"+str(speed), ProductThing))
  21.             mutex.release()
  22.             full.release()
  23.         time.sleep(1)
  24.  
  25.  
  26. def consume(mutex,full,empty,product,pindex,cindex):
  27.     count=0
  28.     time_start=time.clock()
  29.     while True:
  30.         speed=random.randint(1, 5)
  31.         for i in range(speed):
  32.             consumeThing=""
  33.             full.acquire()
  34.             with mutex:
  35.                 consumeThing+=str(product[cindex.value])+' '
  36.                 cindex.value=(cindex.value+1)%len(product)
  37.                 print('%s:  consume things:%s' % ("consumer", consumeThing))
  38.                 count=count+1
  39.                 if(count==20):
  40.                     time_end=time.clock()
  41.                     print("消费 "+str(count)+"个商品所用时间: %f s" % (time_end - time_start))
  42.             empty.release()
  43.         time.sleep(1)
  44.  
  45.  
  46. if __name__ == '__main__':
  47.  
  48.     product = multiprocessing.Array('d',range(5))
  49.  
  50.     pindex=multiprocessing.Value('i',0)
  51.  
  52.     cindex=multiprocessing.Value('i',0)
  53.  
  54.     mutex = multiprocessing.RLock()
  55.  
  56.     full = multiprocessing.Semaphore(0)
  57.  
  58.     empty = multiprocessing.Semaphore(5)
  59.  
  60.     Process(target=produce, args=(1,mutex,full,empty,product,pindex,cindex)).start()
  61.  
  62.     Process(target=produce, args=(2,mutex,full,empty,product,pindex,cindex)).start()
  63.  
  64.     Process(target=consume, args=(mutex,full,empty,product,pindex,cindex)).start()

2、服务进程管理器Manager

  1. '''
  2. @author: jqy
  3. '''
  4.  
  5. from multiprocessing import Process
  6. import time
  7. import random
  8. import multiprocessing
  9.  
  10. def produce(speed,mutex,full,empty,product,pindex,cindex):
  11.     while True:
  12.         for i in range(speed):
  13.             ProductThing='';
  14.             empty.acquire()
  15.             mutex.acquire()
  16.             num=random.random()
  17.             ProductThing+=str(num)+' '
  18.             product[pindex.value]=num
  19.             pindex.value=(pindex.value+1)%len(product)
  20.             print ('%s: product things:%s' % ("producer"+str(speed), ProductThing))
  21.             mutex.release()
  22.             full.release()
  23.         time.sleep(1)
  24.  
  25.  
  26. def consume(mutex,full,empty,product,pindex,cindex):
  27.     count=0
  28.     time_start=time.clock()
  29.     while True:
  30.         speed=random.randint(1, 5)
  31.         for i in range(speed):
  32.             consumeThing=""
  33.             full.acquire()
  34.             with mutex:
  35.                 consumeThing+=str(product[cindex.value])+' '
  36.                 cindex.value=(cindex.value+1)%len(product)
  37.                 print('%s:  consume things:%s' % ("consumer", consumeThing))
  38.                 count=count+1
  39.                 if(count==20):
  40.                     time_end=time.clock()
  41.                     print("消费 "+str(count)+"个商品所用时间: %f s" % (time_end - time_start))
  42.             empty.release()
  43.         time.sleep(1)
  44.  
  45.  
  46. if __name__ == '__main__':
  47.  
  48.     mgr = multiprocessing.Manager()
  49.  
  50.     product = mgr.Array('d',range(5))
  51.  
  52.     pindex = mgr.Value('i',0)
  53.  
  54.     cindex = mgr.Value('i',0)
  55.  
  56.     mutex = mgr.RLock()
  57.  
  58.     full = mgr.Semaphore(0)
  59.  
  60.     empty = mgr.Semaphore(5)
  61.  
  62.     p1 = Process(target=produce, args=(1,mutex,full,empty,product,pindex,cindex))
  63.  
  64.     p2 = Process(target=produce, args=(2,mutex,full,empty,product,pindex,cindex))
  65.  
  66.     p3 = Process(target=consume, args=(mutex,full,empty,product,pindex,cindex))
  67.  
  68.     p1.start(), p2.start(), p3.start()
  69.  
  70.     p1.join(), p2.join(), p3.join()

3、  Condition模型

  1. import multiprocessing 
  2. import random
  3. import time
  4.  
  5. def produce(speed,cond,product):
  6.     while True:
  7.         for i in range(speed): 
  8.             cond.acquire()
  9.             while len(product)>=5:
  10.                 cond.wait();
  11.             num=random.random();
  12.             product.append(num)
  13.             print("生产者"+str(speed)+",生产了："+str(num))
  14.             cond.notify()
  15.             cond.release()
  16.         time.sleep(1)
  17.  
  18. def consume(cond,product):
  19.     count=0
  20.     time_start=time.clock() 
  21.     while True:
  22.         speed=random.randint(1,5);
  23.         for i in range(speed):
  24.             cond.acquire()
  25.             while len(product)<=0:
  26.                 cond.wait()
  27.             num=product[0];
  28.             del product[0]
  29.             print("消费者，消费了"+str(num))
  30.             count=count+1       
  31.             if(count==20):                            
  32.                 time_end=time.clock()        
  33.                 print("消费 "+str(count)+"个商品所用时间: %f s" % (time_end - time_start)) 
  34.             cond.notify()
  35.             cond.release()
  36.         time.sleep(1)
  37.  
  38. if __name__ == '__main__':
  39.     
  40.     c= multiprocessing.Condition();
  41.     
  42.     product=multiprocessing.Manager().list()
  43.        
  44.     p1=multiprocessing.Process(target=produce, args=(1,c,product))
  45.     
  46.     p1.start()  
  47.  
  48.     p2=multiprocessing.Process(target=produce, args=(2,c,product))
  49.     
  50.     p2.start()
  51.     
  52.     c1=multiprocessing.Process(target=consume, args=(c,product))
  53.     
  54.     c1.start()
  55.     
  56.     p1.join()
  57.     
  58.     p2.join()
  59.     
  60.     c1.join()

4、  消息队列模型

  1. '''
  2. @author: jqy
  3. '''
  4. from multiprocessing import Process
  5. import random
  6. import time
  7. import multiprocessing
  8.  
  9. def produce(speed,q):
  10.     while True:
  11.         for i in range(speed):
  12.             item = random.random()
  13.             q.put(item)
  14.             print("生产者",speed,"生产了",str(item))
  15.         time.sleep(1)
  16.  
  17. def consume(num,q):
  18.     count=0
  19.     time_start=time.clock() 
  20.     while True:
  21.         speed=random.randint(1, 5)
  22.         for i in range(speed):
  23.             item = q.get()        
  24.             print("消费者","消费了",item)
  25.             count=count+1       
  26.             if(count==20):                            
  27.                 time_end=time.clock()        
  28.                 print("消费 "+str(count)+"个商品所用时间: %f s" % (time_end - time_start)) 
  29.         time.sleep(1)
  30.  
  31. if __name__ == '__main__':
  32.     
  33.     q =  multiprocessing.Queue(maxsize = 5)
  34.        
  35.     Process(target=produce, args=(1,q)).start()  
  36.  
  37.     Process(target=produce, args=(2,q)).start()
  38.     
  39.     Process(target=consume, args=(1,q)).start()

5、  管道模型：

  1. '''
  2. Created on 2016/11/11
  4. @author: jqy
  5. '''
  6. from multiprocessing import Process
  7. import random
  8. import time
  9. import multiprocessing
  10.  
  11. def produce(speed,empty,p):
  12.     while True:
  13.         for i in range(speed):
  14.             empty.acquire()
  15.             item = random.random()
  16.             p.send(item)
  17.             print("生产者",speed,",生产了",str(item))
  18.         time.sleep(1)
  19.  
  20. def consume(num,empty,p):
  21.     count=0
  22.     time_start=time.clock() 
  23.     while True:
  24.         speed=random.randint(1,5)
  25.         for i in range(speed):
  26.             item = p.recv()        
  27.             print("消费者"," 消费了",item)
  28.             count=count+1       
  29.             if(count==20):                            
  30.                 time_end=time.clock()        
  31.                 print("消费 "+str(count)+"个商品所用时间: %f s" % (time_end - time_start))
  32.             empty.release()
  33.         time.sleep(1)
  34.  
  35. if __name__ == '__main__':
  36.     
  37.     parent_conn, child_conn  =  multiprocessing.Pipe()
  38.  
  39.     empty = multiprocessing.Semaphore(5)
  40.        
  41.     Process(target=produce, args=(1,empty,child_conn)).start()  
  42.  
  43.     Process(target=produce, args=(2,empty,child_conn)).start()
  44.     
  45.     Process(target=consume, args=(1,empty,parent_conn)).start()

6、  Event模型

生产者生产完商品会立即通知消费者去消费，消费者消费完商品后会立即通知生产者去生产，适用于产品池数目为一的情况。

  1. import multiprocessing 
  2. import random
  3. import time
  4.  
  5. def produce(speed,e_p1,e_p2,e_c,product):
  6.     while True:
  7.         for i in range(speed):
  8.             if(speed==1):
  9.                 e_p1.wait();
  10.             if(speed==2):
  11.                 e_p2.wait(); 
  12.             num=random.random();
  13.             product.append(num)
  14.             print("生产者"+str(speed)+",生产了"+str(num))
  15.             e_c.set();
  16.             if(speed==1):
  17.                 e_p1.clear();
  18.             if(speed==2):
  19.                 e_p2.clear(); 
  20.         time.sleep(1)
  21.  
  22. def consume(e_p1,e_p2,e_c,product):
  23.     count=0
  24.     time_start=time.clock() 
  25.     while True:
  26.         speed=random.randint(1,5);
  27.         for i in range(speed):
  28.             e_c.wait();
  29.             num=product[0];
  30.             del product[0]
  31.             print("消费者，消费了"+str(num))
  32.             count=count+1       
  33.             if(count==20):                            
  34.                 time_end=time.clock()        
  35.                 print("消费 "+str(count)+"个商品所用时间: %f s" % (time_end - time_start))
  36.             r=random.randint(1,2)
  37.             if(r==1):
  38.                 e_p1.set()
  39.             if(r==2):
  40.                 e_p2.set() 
  41.             e_c.clear()
  42.         time.sleep(1)
  43.  
  44. if __name__ == '__main__':
  45.     
  46.     e_p1= multiprocessing.Event();
  47.     
  48.     e_p2= multiprocessing.Event();
  49.     
  50.     e_c= multiprocessing.Event();
  51.     
  52.     e_p1.set()
  53.     
  54.     product=multiprocessing.Manager().list()
  55.        
  56.     p1=multiprocessing.Process(target=produce, args=(1,e_p1,e_p2,e_c,product))
  57.     
  58.     p1.start()  
  59.  
  60.     p2=multiprocessing.Process(target=produce, args=(2,e_p1,e_p2,e_c,product))
  61.     
  62.     p2.start()
  63.     
  64.     c1=multiprocessing.Process(target=consume, args=(e_p1,e_p2,e_c,product))
  65.     
  66.     c1.start()
  67.     
  68.     p1.join()
  69.     
  70.     p2.join()
  71.     
  72.     c1.join()

三、不同主机上生产者消费者模型：

1、Manager：服务进程管理器可以通过网络由不同计算机上的进程共享

2、 socketTCP模型：

  1. from multiprocessing import Process
  2. import queue
  3. import threading
  4. import socket
  5. import random
  6. import time
  7.  
  8. MaxSize=5
  9.  
  10. def produce(speed,host,port):
  11.     A=(host,port)
  12.     s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
  13.     s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF,MaxSize*3)
  14.     s.setsockopt(socket.SOL_SOCKET,socket.SO_RCVBUF,MaxSize*3)
  15.     s.connect(A)
  16.     item=1; 
  17.     while True:
  18.         for i in range(speed):
  19.             if(speed==1):
  20.                 print("生产者",speed,",生产了","{0:0=3}".format(item))
  21.                 s.send(bytes("{0:0=3}".format(item), encoding = "utf8"))
  22.             else:
  23.                 print("生产者",speed,",生产了","{0:x=3}".format(item))
  24.                 s.send(bytes("{0:x=3}".format(item), encoding = "utf8"))
  25.             item=item+1
  26.         time.sleep(1)
  27.     s.close()
  28.  
  29. def consume(host1,port1,host2,port2):
  30.     q=queue.Queue(maxsize=MaxSize); 
  31.     threading.Thread(target=getMessage, args=(q,host1,port1)).start()
  32.     threading.Thread(target=getMessage, args=(q,host2,port2)).start()   
  33.     count=0
  34.     time_start=time.clock() 
  35.     while True:
  36.         speed=random.randint(1,MaxSize)
  37.         for i in range(speed):
  38.             item=q.get()
  39.             print("消费者"," 消费了",item)
  40.             count=count+1       
  41.             if(count==20):                            
  42.                 time_end=time.clock()        
  43.                 print("消费"+str(count)+"个商品所用的时间: %f s" % (time_end - time_start))
  44.         time.sleep(1)
  45.  
  46. def getMessage(q,host,port):    
  47.     A=(host,port)
  48.     sock=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
  49.     sock.setsockopt(socket.SOL_SOCKET,socket.SO_RCVBUF,MaxSize*3)
  50.     sock.setsockopt(socket.SOL_SOCKET,socket.SO_SNDBUF,MaxSize*3) 
  51.     sock.bind(A)
  52.     sock.listen(0)
  53.     tcpClientSock, addr=sock.accept()  
  54.     while True:
  55.         try:                  
  56.             data=tcpClientSock.recv(3)
  57.             q.put(str(data,encoding = "utf8"))
  58.             print("count="+str(q.qsize()))        
  59.         except:
  60.             print("exception")
  61.     tcpClientSock.close()
  62.     sock.close()
  63.     
  64.     
  65. if __name__ == '__main__':
  66.        
  67.     Process(target=produce, args=(1,'localhost',8080)).start()  
  68.  
  69.     Process(target=produce, args=(2,'localhost',8090)).start()
  70.     
  71.     Process(target=consume, args=('localhost',8080,'localhost',8090)).start()

3、  远程调用模型：

先在主进程中注册获取产品的方法，消费者在取用商品时调用取用商品的远程方法来获取。取用商品有一定的延迟，使得程序的整个运行速度比较慢。

  1. from multiprocessing import Process
  2. from xmlrpc.client import ServerProxy
  3. from xmlrpc.server import SimpleXMLRPCServer
  4. import random
  5. import time
  6. import multiprocessing
  7.  
  8. q =  multiprocessing.Queue(maxsize = 5)
  9.  
  10. def produce(speed,q):
  11.     while True:
  12.         for i in range(speed):
  13.             item = random.random()
  14.             q.put(item)
  15.             print("生产者",speed,"生产了",str(item))
  16.         time.sleep(1)
  17.  
  18. def getAProduct():
  19.     global q
  20.     return q.get()
  21.  
  22. def consume(host,port):
  23.     server = ServerProxy("http://"+host+":"+str(port))
  24.     count=0
  25.     time_start=time.clock() 
  26.     while True:
  27.         speed=random.randint(1, 5)
  28.         for i in range(speed):
  29.             print("开始远程调用")
  30.             item = server.getAProduct()        
  31.             print("消费者","消费了",item)
  32.             count=count+1       
  33.             if(count==20):                            
  34.                 time_end=time.clock()        
  35.                 print("消费"+str(count)+"个资源所需要的时间: %f s" % (time_end - time_start)) 
  36.         time.sleep(1)
  37.  
  38. if __name__ == '__main__':
  39.     
  40.     s = SimpleXMLRPCServer(('localhost', 8000))
  41.     
  42.     s.register_function(getAProduct)
  43.     
  44.     print('注册获取产品方法完成')
  45.             
  46.     Process(target=produce, args=(1,q)).start()  
  47.  
  48.     Process(target=produce, args=(2,q)).start()
  49.     
  50.     Process(target=consume, args=('localhost',8000)).start()
  51.     
  52.     s.serve_forever()

4、使用Redis发布订阅模式：

  1. #encoding=utf-8
  2. '''
  3. @author: jqy
  4. '''
  5. from multiprocessing import Process
  6. import random
  7. import time
  8. import redis
  9.  
  10. def produce(speed):
  11.     client = redis.StrictRedis()
  12.     while True:
  13.         for i in range(speed):
  14.             item = random.random()
  15.             client.publish("hole",item)
  16.             print("生产者",speed,"生产了",str(item))
  17.         time.sleep(1)
  18.  
  19. def consume(speed):
  20.     count=0
  21.     time_start=time.clock()
  22.     client = redis.StrictRedis()
  23.     p = client.pubsub()
  24.     p.subscribe("hole")
  25.     for i in range(speed):
  26.         for msg in p.listen():
  27.             if msg['type'] != 'message':
  28.                 continue
  29.             print("消费者", "消费了", msg)
  30.             count = count + 1
  31.             if (count == 20):
  32.                 time_end = time.clock()
  33.                 print("消费 " + str(count) + "个商品所用时间: %f s" % (time_end - time_start))
  34.         time.sleep(1)
  35.  
  36. if __name__ == '__main__':
  37.  
  38.     Process(target=consume, args=(2,)).start()
  39.  
  40.     Process(target=produce, args=(1,)).start()
  41.  
  42.     Process(target=produce, args=(2,)).start()

注：如果生产者存在，消费者不存在，那么消息直接会丢弃。生产者发送消息的时候，如果消费者突然断掉了，则消息丢失。

参考文献：

multiprocessing --- 基于进程的并行 — Python 3.10.1 文档

Python使用Manager对象实现不同机器上的进程跨网络传输数据 - 云+社区 - 腾讯云

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/153459
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号