python并发编程之Queue线程、进程、协程通讯（五）

时间 2019-11-12

标签 python 并发编程 queue 线程进程通讯栏目 Python 繁體版

原文原文链接

单线程、多线程之间、进程之间、协程之间不少时候须要协同完成工做，这个时候它们须要进行通信。或者说为了解耦，广泛采用Queue，生产消费模式。html

系列文章

同步deque和多线程Queue

程序有时须要在列表的端点进行操做，比list更加优化的数据结构有Queue和deque。并发

deque

deque通常用在定长队列，多余的数据会被丢弃，这个队列是线程非安全的。app

from queue import Queue, deque

# 大于会截取后面的一段
q = deque(iterable=[1,2,3,4], maxlen=5)
# 参数iterable能够是任何可迭代对象，maxlen表明定长

# 添加与取出
q.append(3) # 从尾部添加
q.pop() # 从尾部弹出一个
q.appendleft(4) # 从首部添加
q.popleft() # 从首部弹出
q.clear() # 清空队列
q.extend([1, 3, 3])  # 将原来的队列从右侧扩展
q.extendleft()  # 将原来的队列从左侧扩展
q.insert(2, 3)  # 在索引为2的位置插入3，若是队列已达到最大，抛出异常

# 复制
q1 = q.copy() # 彻底符合一份队列
# 统计
n = q.count(3) # 统计某个值的数目
x = q.index(3)  # 查找某个值的位置
# 变换
q.reverse() # 将原来的q翻转
q.remove(3) # 删除队列中的全部的3
q.rotate(2) # 向右旋转两步

Queue

Queue提供更加完整成熟的队列操做，相对于deque来讲偏重型，他是线程安全的队列。async

方法和属性分析

from queue import Queue, deque

q = Queue(maxsize=5) #maxsize<=0，队列长度没有限制，这个Queue是线程安全的，经过锁机制保证
print(q.queue) # 一个deque队列
print(q.mutex) # 队列的线程锁
print(q.not_empty) # 非空通知，用在多线程
print(q.not_full) # 非满通知，用在多线程
print(q.all_tasks_done) # 完成的任务
print(q.maxsize)
print(q.unfinished_tasks) # 队列未完成的任务数量，即队列目前的数目

# 数据存取
q.put(3, block=True, timeout=3) # 向队列左边添加数据，block为True队列满了阻塞等待，block为false则直接抛出异常
q.get(block=True, timeout=3) # 队列取出数据，超时抛出异常，block为false忽略timeout
# q.get_nowait() # 当即获取，没有抛出异常
q.put_nowait(4) # 当即插入，已满抛出异常
# 判断
q.full() # 判断当前队列是否已满，满了返回True
q.empty() # 判断当前队列是否为空，空返回True

# 统计
q.task_done() # 用来通知队列任务完成
q.qsize() # 当前队列的任务数量，不绝对可靠

q.join() # 阻塞直到全部的任务完成，即q.unfinished_tasks降为0

实例

from threading import Thread
from queue import Queue, deque
import time

def get_from_queue(queue:Queue):
    while True:
        if not queue.empty():
            print(queue.get_nowait())
            queue.task_done() # 任务完成

def put_to_queue(queue:Queue):
    for i in range(100):
        if not queue.full():
            queue.put_nowait(i)
        else:
            time.sleep(0.1)

q = Queue(5)
th1 = Thread(target=get_from_queue, args=(q,))
th2 = Thread(target=put_to_queue, args=(q,))
th1.start()
th2.start()

进程间通信

multiprocessing的Queue对象能够做为进程间通信的第三者。

from multiprocessing import Queue, Process, Pool
import time

def get_from_queue(queue:Queue):
    while True:
        if not queue.empty():
            print(queue.get_nowait())

def put_to_queue(queue:Queue):
    for i in range(100):
        if not queue.full():
            queue.put_nowait(i)
        else:
            time.sleep(0.1)
if __name__ == '__main__':
    q = Queue(9) # 这个Queue能够在多个进程之间共享
    p1 = Process(target=get_from_queue, args=(q,))
    p2 = Process(target=put_to_queue, args=(q,))
    p1.start()
    p2.start()

multiprocessing.Queue对象

Queue对象的大部分方法和Queue.Queue的方法相同，用法也同样，但有几个特殊的方法:

q = Queue(9) # 这个Queue能够在多个进程之间共享
# q.close() # 关闭队列，再也不接收数据
# q.cancel_join_thread() # 取消阻塞等待
q.join_thread() # 线程阻塞等待

gevent协程的Queue

gevent.queue.Queue基于协程，Queue在多个协程间共享，Queue实现了迭代器协议，可使用for循环遍历。

from gevent.queue import Queue
import gevent
import time

def get_from_queue(queue:Queue, n):
    i = 0
    print('start---get--{}'.format(n))
    while True:
        print(str(queue.get()) + 'get' + str(n))
        i += 1
        if i == 100:
            break

def put_to_queue(queue:Queue, n):
    i = 0
    print('start---put--{}'.format(n))
    while True:
        queue.put(i)
        print(str(i) + 'put' + str(n))
        i += 1
        if i == 100:
            break
if __name__ == '__main__':
    q = Queue(9) # 这个Queue能够在多个进程之间共享
    job1 = [gevent.spawn(put_to_queue, q,i) for i in range(2)]
    job2 = [gevent.spawn(get_from_queue, q,i) for i in range(2)]
    job1.extend(job2)
    gevent.joinall(job1)

协程启动后会按照添加到循环的顺序开始执行，上例在队列未满以前一直执行put操做，直到队列满后阻塞就切换到put2协程，也会当即阻塞，而后切换到get1协程，获取全部的值直到队列为空后阻塞切换。

gevent.queue.Queue对象

其方法基本和Queue.Queue的方法相同，特殊方法以下：

q = Queue(9, items=[1,2,3, StopIteration]) # 实现迭代协议，最后一个必须是StopIteration
# q.copy() #复制一个队列
x = q.next() # 唤醒获取值
q.peek(block=True, timeout=None) # 获取一个值可是不删除它
q.peek_nowait() # 当即获取，忽略timeout
q.put() # 会唤醒多个协程完成添加操做
q.get() # 会挂起多个协程

gevent.queue.JoinableQueue对象扩展了Queue的功能，添加了task_done和join方法。

q = JoinableQueue(9, items=[1,2,3, StopIteration]) # 这个Queue能够在多个进程之间共享
q.task_done() # 通知队列一个任务完成
q.unfinished_tasks # 未完成的任务计数
q.join() # 阻塞等待任务完成，若是unfinished_tasks降为0，则解除

实例

from gevent.queue import Queue, JoinableQueue
import gevent
import time

def get_from_queue(queue:JoinableQueue):
    while True:
        try:
            x = queue.get() # 阻塞时就会切换协程
            print(x)
        finally:
            queue.task_done()

if __name__ == '__main__':
    q = JoinableQueue(8)
    job1 = [gevent.spawn(get_from_queue, q) for i in range(2)]
    for i in range(100):
        q.put(i) # 当Put被阻塞时将切换协程,
    q.join() # 若是不等待的话，最后一次put后将直接退出

参考
http://www.gevent.org/api/gevent.queue.html