使用Python队列和多线程实现生产者消费者

这个模型使用python来实现相比POSIX来做简单太多太多了,轮子python都可以写好了直接调用即可,队列就已经封装好了对共享数据的安全访问。在POSIX多线程中考虑Mutex和条件变量是一个重点。这是我以前用POSIX pthread函数实现的一个生产者和消费者模型:

一、实现方式

下面是Python实现的方法,实际上就是使用队列作为共享数据的中间价这是模块封装好的,不需要在使用Mutex保护线程之间共享数据访问的
安全性。附带队列的测试代码,代码简单如下:


点击(此处)折叠或打开

  1. import threading
  2. import time
  3. import queue
  4. from global_par import global_par
  5. #queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.

  6. def Prod(global_data):
  7.     """生产者线程"""
  8.     i = 1
  9.     qfifo = global_data.get_global_pars("CusQue")
  10.     while 1:
  11.         qfifo.put(i)
  12.         print("生产产品: {} ".format(i))
  13.         i +=1;
  14.     return 0

  15. def Cust(global_data,detail):
  16.     """消费者线程"""
  17.     qfifo = global_data.get_global_pars("CusQue")
  18.     while 1:
  19.         print("{}:消费产品:{}".format(detail,qfifo.get()))
  20.         time.sleep(1)
  21.     return 0

  22. def main():
  23.     pthread_list = []
  24.     global_data = global_par()
  25.     qfifo = queue.Queue(maxsize=10)
  26.     global_data.set_global_pars("CusQue",qfifo)

  27.     ##建立1个生产者线程
  28.     thread_pro = threading.Thread(target=Prod,args=(global_data,))
  29.     pthread_list.append(thread_pro)

  30.     ##建立10个消费者线程
  31.     for i in range(10):
  32.         thread_cus = threading.Thread(target=Cust, args=(global_data, "消费线程{}".format(i+1),))
  33.         pthread_list.append(thread_cus)

  34.     for i in pthread_list:
  35.         i.start()

  36.     for i in pthread_list:
  37.         i.join

  38.     return 0

  39. if __name__ == '__main__':
  40.     main()

二、队列测试代码


点击(此处)折叠或打开

  1. import threading
  2. import queue
  3. #http://www.cnblogs.com/alex3714/articles/5230609.html

  4. def my_format(x, y):
  5.     """par one: what des your print
  6.        par two: value
  7.     """
  8.     x = "-->" + x + " {}"
  9.     print(x.format(y))
  10.     return

  11. qfifo = queue.Queue(maxsize=2) ##先进先出队列 如果达到最大值默认Queue.put()会堵塞 如果为空默认Queue.get()会堵塞
  12. qlifo = queue.LifoQueue(maxsize=2) ##后入先出队列 如果达到最大值默认Queue.put()会堵塞 如果为空默认Queue.get()会堵塞
  13. qpri = queue.PriorityQueue(maxsize=2) ##根据优先级判断顺序小的为优先级高 如果达到最大值默认Queue.put()会堵塞 如果为空默认Queue.get()会堵塞

  14. ##part0:在队列增加元素Queue.put(item, block=True, timeout=None) 可以设置是否堵塞
  15. my_format("##part0:在队列增加元素Queue.put(item, block=True, timeout=None):","")
  16. qfifo.put(1)
  17. qfifo.put(2)
  18. ##part1:队列满了会报queue.Full异常
  19. try:
  20.     qfifo.put(2,timeout=1)
  21. except queue.Full as e:
  22.     my_format("##part1:队列满了如果不堵塞会报queue.Full异常", "")

  23. ##part2:出队列Queue.get(block=True, timeout=None) 可以设置是否堵塞

  24. my_format("##part2:出队列Queue.get(block=True, timeout=None) 可以设置是否堵塞:","")
  25. qfifo.get()
  26. qfifo.get()

  27. ##part3:队列空如果不堵塞会报异常queue.Empty
  28. try:
  29.     qfifo.get(timeout=1)
  30. except queue.Empty as e:
  31.     my_format("##part3:队列空如果不堵塞会报异常queue.Empty", "")


  32. ##part4:返回队列的大小Queue.qsize()
  33. my_format("##part0:返回队列的大小Queue.qsize():",qfifo.qsize())

  34. ##part5:返回是否为空的bool值Queue.empty()
  35. my_format("##part1:返回是否为空的bool值Queue.empty():",qfifo.empty())

  36. ##part6:返回是否满的bool值Queue.full()

  37. my_format("##part1:返回是否为空的bool值Queue.full():",qfifo.full())

作者微信:




请使用浏览器的分享功能分享到微信等