解析python 生产/消费者模型实现过程

1.多线程实现

import threading
import queue
import logging
logging.basicConfig(level = logging.INFO,format = '%(asctime)s  - %(levelname)s -->%(funcName)s  at line %(lineno)d: \n %(message)s')
log= logging.getLogger()

def  producer(q):
    for k in range(10):
        q.put(k)
        logging.info("put %s item into queue "%k)
    # block all producer main until consumer get all queue members that put
    q.join()

def consumer(q):
    while True:
        """
            if queue been empty and q.get(block=True),
        because default block in source code is True ,
        queue will bock forever util queue put new item .
        """
        item=q.get()
        logging.info("get out of %s member of queue "%item)

        # notify q.join() consumer has get out of member of queue
        q.task_done()

if __name__ == '__main__':
    thread_num=5
    # limit max length of queue
    q = queue.Queue(maxsize=5)
    producers=[threading.Thread(target=producer,args=(q,))]
    consumers=[threading.Thread(target=consumer,args=(q,))  for i in range(thread_num)]
    # start producer and  consumer
    for pr in producers:
        pr.start()
    for cn in consumers:
        cn.start()
    # block main thread of producer
    for p  in producers:
        p.join()
    # block main of consumer
    for c in consumers:
        c.join()

执行结果：可以看到控制台一直阻塞在那里。由于消费者消费完queue里面所有的成员后queue 变成empty，空队列，由于源码q.get()内置参数block一直为True，当取不到时一直阻塞，那么这就意味着当你block 为True时必须设置结束标志让消费者结束，退出死循环接下来我们
改良一下这个方法：

在生产者生产的时候加入结束标志None告诉消费者拿到None时就break 跳出死循环，结束消费，这样就解决了消费之一直阻塞问题：

在producer主线程join之前添加与线程数目相同的None，让每个thread break 跳出死循环：

import threading

import queue

import logging

logging.basicConfig(level = logging.INFO,format = '%(asctime)s  - %(levelname)s -->%(funcName)s  at line %(lineno)d: \n %(message)s')

log= logging.getLogger()

def  producer(q):

    for k in range(10):

        q.put(k)

        logging.info("put %s item into queue "%k)

    # block all producer main until consumer get all queue members that put

    q.join()

def consumer(q):

    while True:

        """

            if queue been empty and q.get(block=True),

        because default block in source code is True ,

        queue will bock forever util queue put new item .

        """

        item=q.get()

        logging.info("get out of %s member of queue " % item)

        if item == None:

            logging.info("get None stop consumer from while loop")

            break

        # notify q.join() consumer has get out of member of queue

        q.task_done()

if __name__ == '__main__':

    thread_num=5

    # limit max length of queue

    q = queue.Queue(maxsize=5)

    producers=[threading.Thread(target=producer,args=(q,))]

    consumers=[threading.Thread(target=consumer,args=(q,))  for i in range(thread_num)]

    # start producer and  consumer

    for pr in producers:

        pr.start()

    for cn in consumers:

        cn.start()

    # block main thread of producer

    for p  in producers:

        p.join()

    # put same number of thread_num None to break thread from consumer while

    for i in range(thread_num):

        q.put(None)

    # block main of consumer

    for c in consumers:

        c.join()

2019-12-21 20:37:00,206 - INFO -->consumer at line 23:
get out of None member of queue
2019-12-21 20:37:00,206 - INFO -->consumer at line 25:
get None mean break
2019-12-21 20:37:00,206 - INFO -->consumer at line 25:
get None mean break
2019-12-21 20:37:00,206 - INFO -->consumer at line 25:
get None mean break
2019-12-21 20:37:00,206 - INFO -->consumer at line 25:
get None mean break
2019-12-21 20:37:00,206 - INFO -->consumer at line 23:
get out of None member of queue
2019-12-21 20:37:00,207 - INFO -->consumer at line 25:
get None mean break

Process finished with exit code 0

结果：正常释放所有线程，但是好像每次线程都要加个标志太麻烦，感觉，能不能有更好的办法呢，当然接着往下看：

改良consumer方法在每次消费队列之前检查队列queue.empty() 是否为True,是就直接break跳出while loop:就不需要每次都加None标志了

import threading

import queue

import logging

logging.basicConfig(level = logging.INFO,format = '%(asctime)s  - %(levelname)s -->%(funcName)s  at line %(lineno)d: \n %(message)s')

log= logging.getLogger()

def  producer(q):

    for k in range(10):

        q.put(k)

        logging.info("put %s item into queue "%k)

    # block all producer main until consumer get all queue members that put

    q.join()

def consumer(q):

    while True:

        """

            if queue been empty and q.get(block=True),

        because default block in source code is True ,

        queue will bock forever util queue put new item .

        """

        if q.empty():break

        item=q.get()

        logging.info("get out of %s member of queue " % item)

        # notify q.join() consumer has get out of member of queue

        q.task_done()

if __name__ == '__main__':

    thread_num=5

    # limit max length of queue

    q = queue.Queue(maxsize=5)

    producers=[threading.Thread(target=producer,args=(q,))]

    consumers=[threading.Thread(target=consumer,args=(q,))  for i in range(thread_num)]

    # start producer and  consumer

    for pr in producers:

        pr.start()

    for cn in consumers:

        cn.start()

    # block main thread of producer

    for p  in producers:

        p.join()

j结果如下：

巴特西

解析python 生产/消费者模型实现过程

最新文章

热门文章