python standard library queue

发表于 | 更新于 | 分类于 ， | 阅读次数：
本文字数： 7.6k | 阅读时长 ≈ 7 分钟

python 标准库 queue

源码

源代码: Lib/queue.py

queue 模块实现了多生产者、多消费者队列。这特别适用于消息必须安全地在多线程间交换的线程编程。模块中的 Queue 类实现了所有所需的锁定语义。

模块实现了三种类型的队列，它们的区别仅仅是条目取回的顺序。在 FIFO 队列中，先添加的任务先取回。在 LIFO 队列中，最近被添加的条目先取回(操作类似一个堆栈)。优先级队列中，条目将保持排序( 使用 heapq 模块 ) 并且最小值的条目第一个返回。

在内部，这三个类型的队列使用锁来临时阻塞竞争线程；然而，它们并未被设计用于线程的重入性处理。

此外，模块实现了一个 “简单的” FIFO 队列类型， SimpleQueue ，这个特殊实现为小功能在交换中提供额外的保障。

queue 模块定义了下列类和异常：

  • class queue.Queue(maxsize=0)

    Constructor for a FIFO queue. maxsize is an integer that sets the upperbound limit on the number of items that can be placed in the queue. Insertion will block once this size has been reached, until queue items are consumed. If maxsize is less than or equal to zero, the queue size is infinite.

  • class queue.LifoQueue(maxsize=0)

    LIFO 队列构造函数。 maxsize 是个整数，用于设置可以放入队列中的项目数的上限。当达到这个大小的时候，插入操作将阻塞至队列中的项目被消费掉。如果 maxsize 小于等于零，队列尺寸为无限大。

  • class queue.PriorityQueue(maxsize=0)

    优先级队列构造函数。 maxsize 是个整数，用于设置可以放入队列中的项目数的上限。当达到这个大小的时候，插入操作将阻塞至队列中的项目被消费掉。如果 maxsize 小于等于零，队列尺寸为无限大。最小值先被取出( 最小值条目是由 sorted(list(entries))[0] 返回的条目)。条目的典型模式是一个以下形式的元组： (priority_number, data) 。如果 data 元素没有可比性，数据将被包装在一个类中，忽略数据值，仅仅比较优先级数字 ：from dataclasses import dataclass, field from typing import Any @dataclass(order=True) class PrioritizedItem: priority: int item: Any=field(compare=False)

  • class queue.SimpleQueue

    无界的 FIFO 队列构造函数。简单的队列，缺少任务跟踪等高级功能。3.7 新版功能.

  • exception queue.Empty

    对空的 Queue 对象，调用非阻塞的 get() (or get_nowait()) 时，引发的异常。

  • exception queue.Full

    对满的 Queue 对象，调用非阻塞的 put() (or put_nowait()) 时，引发的异常。

Queue对象

队列对象 (Queue, LifoQueue, 或者 PriorityQueue) 提供下列描述的公共方法。

  • Queue.qsize()

    返回队列的大致大小。注意，qsize() > 0 不保证后续的 get() 不被阻塞，qsize() < maxsize 也不保证 put() 不被阻塞。

  • Queue.empty()

    如果队列为空，返回 True ，否则返回 False 。如果 empty() 返回 True ，不保证后续调用的 put() 不被阻塞。类似的，如果 empty() 返回 False ，也不保证后续调用的 get() 不被阻塞。

  • Queue.full()

    如果队列是满的返回 True ，否则返回 False 。如果 full() 返回 True 不保证后续调用的 get() 不被阻塞。类似的，如果 full() 返回 False 也不保证后续调用的 put() 不被阻塞。

  • Queue.put(item, block=True, timeout=None)

    将 item 放入队列。如果可选参数 block 是 true 并且 timeout 是 None (默认)，则在必要时阻塞至有空闲插槽可用。如果 timeout 是个正数，将最多阻塞 timeout 秒，如果在这段时间没有可用的空闲插槽，将引发 Full 异常。反之 (block 是 false)，如果空闲插槽立即可用，则把 item 放入队列，否则引发 Full 异常 ( 在这种情况下，timeout 将被忽略)。

  • Queue.put_nowait(item)

    相当于 put(item, False) 。

  • Queue.get(block=True, timeout=None)

    从队列中移除并返回一个项目。如果可选参数 block 是 true 并且 timeout 是 None (默认值)，则在必要时阻塞至项目可得到。如果 timeout 是个正数，将最多阻塞 timeout 秒，如果在这段时间内项目不能得到，将引发 Empty 异常。反之 (block 是 false) , 如果一个项目立即可得到，则返回一个项目，否则引发 Empty 异常 (这种情况下，timeout 将被忽略)。POSIX系统3.0之前，以及所有版本的Windows系统中，如果 block 是 true 并且 timeout 是 None ， 这个操作将进入基础锁的不间断等待。这意味着，没有异常能发生，尤其是 SIGINT 将不会触发 KeyboardInterrupt 异常。

  • Queue.get_nowait()

    相当于 get(False) 。

提供了两个方法，用于支持跟踪 排队的任务 是否 被守护的消费者线程 完整的处理。

  • Queue.task_done()

    表示前面排队的任务已经被完成。被队列的消费者线程使用。每个 get() 被用于获取一个任务， 后续调用 task_done() 告诉队列，该任务的处理已经完成。如果 join() 当前正在阻塞，在所有条目都被处理后，将解除阻塞(意味着每个 put() 进队列的条目的 task_done() 都被收到)。如果被调用的次数多于放入队列中的项目数量，将引发 ValueError 异常 。

  • Queue.join()

    阻塞至队列中所有的元素都被接收和处理完毕。当条目添加到队列的时候，未完成任务的计数就会增加。每当消费者线程调用 task_done() 表示这个条目已经被回收，该条目所有工作已经完成，未完成计数就会减少。当未完成计数降到零的时候， join() 阻塞被解除。

如何等待排队的任务被完成的示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def worker():
    while True:
        item = q.get()
        if item is None:
            break
        do_work(item)
        q.task_done()

q = queue.Queue()
threads = []
for i in range(num_worker_threads):
    t = threading.Thread(target=worker)
    t.start()
    threads.append(t)

for item in source():
    q.put(item)

# block until all tasks are done
q.join()

# stop workers
for i in range(num_worker_threads):
    q.put(None)
for t in threads:
    t.join()

SimpleQueue 对象

SimpleQueue 对象提供下列描述的公共方法。

  • SimpleQueue.qsize()

    返回队列的大致大小。注意，qsize() > 0 不保证后续的 get() 不被阻塞。

  • SimpleQueue.empty()

    如果队列为空，返回 True ，否则返回 False 。如果 empty() 返回 False ，不保证后续调用的 get() 不被阻塞。

  • SimpleQueue.put(item, block=True, timeout=None)

    将 item 放入队列。此方法永不阻塞，始终成功（除了潜在的低级错误，例如内存分配失败）。可选参数 block 和 timeout 仅仅是为了保持 Queue.put() 的兼容性而提供，其值被忽略。CPython implementation detail: This method has a C implementation which is reentrant. That is, a put() or get() call can be interrupted by another put() call in the same thread without deadlocking or corrupting internal state inside the queue. This makes it appropriate for use in destructors such as __del__ methods or weakref callbacks.

  • SimpleQueue.put_nowait(item)

    相当于 put(item) ，仅为保持 Queue.put_nowait() 兼容性而提供。

  • SimpleQueue.get(block=True, timeout=None)

    从队列中移除并返回一个项目。如果可选参数 block 是 true 并且 timeout 是 None (默认值)，则在必要时阻塞至项目可得到。如果 timeout 是个正数，将最多阻塞 timeout 秒，如果在这段时间内项目不能得到，将引发 Empty 异常。反之 (block 是 false) , 如果一个项目立即可得到，则返回一个项目，否则引发 Empty 异常 (这种情况下，timeout 将被忽略)。

  • SimpleQueue.get_nowait()

    相当于 get(False) 。

小结

Queue的种类：

  • FIFO：

    Queue.Queue(maxsize=0)

FIFO即First in First Out,先进先出。Queue提供了一个基本的FIFO容器，使用方法很简单,maxsize是个整数，指明了队列中能存放的数据个数的上限。一旦达到上限，插入会导致阻塞，直到队列中的数据被消费掉。如果maxsize小于或者等于0，队列大小没有限制。

  • LIFO

Queue.LifoQueue(maxsize=0)

LIFO即Last in First Out,后进先出。与栈的类似，使用也很简单,maxsize用法同上

  • priority

class Queue.PriorityQueue(maxsize=0)

构造一个优先队列。maxsize用法同上。

例子1：LifoQueue

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
import queue
import threading
import time
 
# 可以设置队列的长度 q=queue.LifoQueue(5)，意味着队列中最多存放5个元素,当队列满的时候自动进入阻塞状态
q=queue.LifoQueue()
def put():
    for i in range(10):
        q.put(i)
        print("数据%d被存入到队列中" % i)
    q.join()
    print('ok')
 
def get():
    for i in range(10):
        value = q.get()
        print("数据%d从队列中取出" % value)
        q.task_done()
 
t1=threading.Thread(target=put,args=())
t1.start()
t2=threading.Thread(target=get,args=())
t2.start()

>>> 数据0被存入到队列中
>>> 数据1被存入到数据中
>>> 数据2被存入到队列中
>>> 数据3被存入到队列中
>>> 数据4被存入到队列中
>>> 数据5被存入到队列中
>>> 数据6被存入到队列中
>>> 数据7被存入到队列中
>>> 数据8被存入到队列中
>>> 数据9被存入到队列中
>>> 数据9从队列中取出
>>> 数据8从队列中取出
>>> 数据7从队列中取出
>>> 数据6从队列中取出
>>> 数据5从队列中取出
>>> 数据4从队列中取出
>>> 数据3从队列中取出
>>> 数据2从队列中取出
>>> 数据1从队列中取出
>>> 数据0从队列中取出
>>> ok

例子2：Priortity Queue in Python

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
# A simple implementation of Priority Queue 
# using Queue. 
class PriorityQueue(object): 
    def __init__(self): 
        self.queue = [] 
  
    def __str__(self): 
        return ' '.join([str(i) for i in self.queue]) 
  
    # for checking if the queue is empty 
    def isEmpty(self): 
        return len(self.queue) == [] 
  
    # for inserting an element in the queue 
    def insert(self, data): 
        self.queue.append(data) 
  
    # for popping an element based on Priority 
    def delete(self): 
        try: 
            max = 0
            for i in range(len(self.queue)): 
                if self.queue[i] > self.queue[max]: 
                    max = i 
            item = self.queue[max] 
            del self.queue[max] 
            return item 
        except IndexError: 
            print() 
            exit() 
  
if __name__ == '__main__': 
    myQueue = PriorityQueue() 
    myQueue.insert(12) 
    myQueue.insert(1) 
    myQueue.insert(14) 
    myQueue.insert(7) 
    print(myQueue)             
    while not myQueue.isEmpty(): 
        print(myQueue.delete())  

>>> 12 1 14 7
>>> 14
>>> 12
>>> 7
>>> 1
>>> ()

扩展:第三方队列

下面介绍五个第三方队列框架,看来造轮子也是个好方法，:)

  1. Celery

    官方栗子

  2. RQ

一个栗子

1
2
3
4
5
import requests

def count_words_at_url(url):
    resp = requests.get(url)
    return len(resp.text.split())

创建一个RQ queue

1
2
3
4
from redis import Redis
from rq import Queue

q = Queue(connection=Redis())

方法调用

1
2
3
from my_module import count_words_at_url
result = q.enqueue(
             count_words_at_url, 'http://nvie.com')

Worker实例

1
2
3
4
5
$ rq worker
*** Listening for work on default
Got count_words_at_url('http://nvie.com') from default
Job result = 818
*** Listening for work on default

  1. huey

    一个轻量级队列框架

    一个栗子

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    from huey import RedisHuey, crontab

    huey = RedisHuey('my-app', host='redis.myapp.com')

    @huey.task()
    def add_numbers(a, b):
        return a + b

    @huey.task(retries=2, retry_delay=60)
    def flaky_task(url):
        # This task might fail, in which case it will be retried up to 2 times
        # with a delay of 60s between retries.
        return this_might_fail(url)

    @huey.periodic_task(crontab(minute='0', hour='3'))
    def nightly_backup():
        sync_all_data()

  1. kuyruk

    创建Task

    1
    2
    3
    4
    5
    6
    7
    8
    # tasks.py
    from kuyruk import Kuyruk

    kuyruk = Kuyruk()

    @kuyruk.task()
    def echo(message):
        print message

    发送Task 去 RabbitMQ

    1
    2
    import tasks
    tasks.echo("Hello, World!")

    运行Worker

    1
    kuyruk --app tasks.kuyruk worker

  2. Dramatiq

    一个栗子

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    import dramatiq
    import requests


    @dramatiq.actor
    def count_words(url):
         response = requests.get(url)
         count = len(response.text.split(" "))
         print(f"There are {count} words at {url!r}.")


    # Synchronously count the words on example.com in the current process
    count_words("http://example.com")

    # or send the actor a message so that it may perform the count
    # later, in a separate process.
    count_words.send("http://example.com")
坚持原创技术分享，您的支持将鼓励我继续创作！
0%