一文速通Python并行计算：11 Python多进程编程-进程之间的数据安全传输-基于队列和管道

方法	作用
Put()	用以插入数据到队列中，put 方法还有两个可选参数：blocked 和 timeout。① 如果 blocked 为 True（默认值），并且 timeout 为正值，该方法会阻塞 timeout 指定的时间，直到该队列有剩余的空间。如果超时，会抛出 Queue.Full 异常。② 如果 blocked 为 False，但该 Queue 已满，会立即抛出 Queue.Full 异常。
get()	可以从队列读取并且删除一个元素。同样，get 方法有两个可选参数：blocked 和 timeout。① 如果 blocked 为 True（默认值），并且 timeout 为正值，那么在等待时间内没有取到任何元素，会抛出 Queue.Empty 异常。② 如果 blocked 为 False，有两种情况存在，如果 Queue 有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出 Queue.Empty 异常.
get_nowait()	同 q.get(False)
put_nowait()	同 q.put(False)
empty()	调用此方法时 q 为空则返回 True，该结果不可靠，比如在返回 True 的过程中，如果队列中又加入了项目。
full()	调用此方法时 q 已满则返回 True，该结果不可靠，比如在返回 True 的过程中，如果队列中的项目被取走。
qsize()	返回队列中目前项目的正确数量，结果也不可靠，理由同 q.empty()和 q.full()一样。
cancel_join_thread()	不会在进程退出时自动连接后台线程。可以防止 join_thread()方法阻塞
close()	关闭队列，防止队列中加入更多数据。调用此方法，后台线程将继续写入那些已经入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果 q 被垃圾收集，将调用此方法。关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如，如果某个使用者正在被阻塞在 get()操作上，关闭生产者中的队列不会导致 get()方法返回错误。
join_thread()	连接队列的后台线程。此方法用于在调用 q.close()方法之后，等待所有队列项被消耗。默认情况下，此方法由不是 q 的原始创建者的所有进程调用。调用 q.cancel_join_thread 方法可以禁止这种行为

在下面的例子中， Producer 类生产 item 放到队列中，然后 Consumer 类从队列中移除它们。

import multiprocessing
import random
import time

class Producer(multiprocessing.Process):
    def __init__(self, queue):
        multiprocessing.Process.__init__(self)
        self.queue = queue

    def run(self):
        for i in range(10):
            item = random.randint(0, 256)
            self.queue.put(item)
            print("Process Producer : item %d appended to queue %s" % (item, self.name))
            time.sleep(1)
            print("The size of queue is %s" % self.queue.qsize())

class Consumer(multiprocessing.Process):
    def __init__(self, queue):
        multiprocessing.Process.__init__(self)
        self.queue = queue

    def run(self):
        while True:
            if self.queue.empty():
                print("the queue is empty")
                break
            else:
                time.sleep(2)
                item = self.queue.get()
                print('Process Consumer : item %d popped from by %s n' % (item, self.name))
                time.sleep(1)

if __name__ == '__main__':
    queue = multiprocessing.Queue()
    process_producer = Producer(queue)
    process_consumer = Consumer(queue)
    process_producer.start()
    process_consumer.start()
    process_producer.join()
    process_consumer.join()

此代码的缺点是在慢生产快消费的情况下主进程会提前结束，其中一种解决方式是让生产者在生产完毕后，往队列中再发一个结束信号，这样消费者在接收到结束信号后就可以 break 出死循环。

但是又出现一个新问题，在有多个生产者和多个消费者时，有几个消费者就需要发送几次结束信号,十分麻烦。其实我们的思路无非是发送结束信号而已，有另外一种队列提供了这种机制。

队列还有一个 JoinableQueue 子类，它有以下两个额外的方法：

**task_done()****: **消费者调用此方法表示之前入队的一个任务已经完成，比如 get() 方法从队列取回 item 之后调用，表示 q.get() 的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发 ValueError 异常。
**join()****: **生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用 task_done() 方法为止。

示例代码如下：

from multiprocessing import Process,JoinableQueue
import time,random,os
def consumer(q):
    while True:
        res=q.get()
        time.sleep(random.randint(1,3))
        print('

登录查看全部

参与评论

一文速通Python并行计算：11 Python多进程编程-进程之间的数据安全传输-基于队列和管道

一文速通 Python 并行计算：11 Python 多进程编程-进程之间的数据安全传输-基于队列和管道

摘要：

原文链接：

往期推荐：

更多精彩内容可看：

文档获取：

正文

1.使用队列进行通信

参与评论

评论留言

还没有评论留言，赶紧来抢楼吧~~

给这篇文章打个标签吧~