python 实现消费者优先级队列

本文分享自天翼云开发者社区《python 实现消费者优先级队列》，作者：Frost
关键字
条件变量，信号量，消费者优先级，公平性，堆队列算法
需求背景
常见的是消息队列支持为消息指定优先级，但支持为消费者指定优先级的却很少见，作者在网上检索一般能查到 rabbitMQ 的消费者优先级相关资料。并没有找到别的语言的资料。
而 python 标准库里全部队列都是公平的，并没有使用非公平的参数，因此大概不能满足有些场景的需求。
什么是公平与非公平呢，这个一般是指互斥锁的特征，互斥锁的多个尝试取锁的线程其实很类似队列的多个消费者，以 waiter 统称。
假设有 A, B, C, D 四个 waiter，他们按照字母顺序依次调用 acquire()/get()，
那么等到有线程开释锁或队列放入了一条消息，会按照先来后到的顺序，唤醒对应的 waiter，也就是这里的 A，同理，按照排队顺序，B -> C -> D 将是后续的唤醒顺序，其实简单讲就是 FIFO。
一般来说 FIFO 策略具有普适性，可以避免有的消费者被饿死，但某些场景我们希望给队列的消费者赋予优先级，每次优先唤醒仍在等待消费的优先级最高的消费者。
下面会给出 pure python 的实现。
实现原理
先阅读 python 自带的 SimpleQueue 源码 (pure python 版本，位于 Lib\queue.py）。

1. class _PySimpleQueue:
2.     '''Simple, unbounded FIFO queue.
4.     This pure Python implementation is not reentrant.
5.     '''
6.     # Note: while this pure Python version provides fairness
7.     # (by using a threading.Semaphore which is itself fair, being based
8.     #  on threading.Condition), fairness is not part of the API contract.
9.     # This allows the C version to use a different implementation.
11.     def __init__(self):
12.         self._queue = deque()
13.         self._count = threading.Semaphore(0)
15.     def put(self, item, block=True, timeout=None):
16.         '''Put the item on the queue.
18.         The optional 'block' and 'timeout' arguments are ignored, as this method
19.         never blocks.  They are provided for compatibility with the Queue class.
20.         '''
21.         self._queue.append(item)
22.         self._count.release()
24.     def get(self, block=True, timeout=None):
25.         '''Remove and return an item from the queue.
27.         If optional args 'block' is true and 'timeout' is None (the default),
28.         block if necessary until an item is available. If 'timeout' is
29.         a non-negative number, it blocks at most 'timeout' seconds and raises
30.         the Empty exception if no item was available within that time.
31.         Otherwise ('block' is false), return an item if one is immediately
32.         available, else raise the Empty exception ('timeout' is ignored
33.         in that case).
34.         '''
35.         if timeout is not None and timeout < 0:
36.             raise ValueError("'timeout' must be a non-negative number")
37.         if not self._count.acquire(block, timeout):
38.             raise Empty
39.         return self._queue.popleft()
41.     def put_nowait(self, item):
42.         '''Put an item into the queue without blocking.
44.         This is exactly equivalent to `put(item, block=False)` and is only provided
45.         for compatibility with the Queue class.
46.         '''
47.         return self.put(item, block=False)
49.     def get_nowait(self):
50.         '''Remove and return an item from the queue without blocking.
52.         Only get an item if one is immediately available. Otherwise
53.         raise the Empty exception.
54.         '''
55.         return self.get(block=False)
57.     def empty(self):
58.         '''Return True if the queue is empty, False otherwise (not reliable!).'''
59.         return len(self._queue) == 0
61.     def qsize(self):
62.         '''Return the approximate size of the queue (not reliable!).'''
63.         return len(self._queue)
65.     __class_getitem__ = classmethod(types.GenericAlias)

复制代码

docstring 里面说明，这个队列是保证了公平性，因为其使用的信号量实现是公平的。
符合直觉的是，我们在 get 方法以及信号量的 acquire 方法增加一个优先级数值的参数，那么再来看信号量的实现，看看能不能做到这一点，

1. class Semaphore:
2.     """This class implements semaphore objects.
4.     Semaphores manage a counter representing the number of release() calls minus
5.     the number of acquire() calls, plus an initial value. The acquire() method
6.     blocks if necessary until it can return without making the counter
7.     negative. If not given, value defaults to 1.
9.     """
11.     # After Tim Peters' semaphore class, but not quite the same (no maximum)
13.     def __init__(self, value=1):
14.         if value < 0:
15.             raise ValueError("semaphore initial value must be >= 0")
16.         self._cond = Condition(Lock())
17.         self._value = value
19.     def acquire(self, blocking=True, timeout=None):
20.         """Acquire a semaphore, decrementing the internal counter by one.
22.         When invoked without arguments: if the internal counter is larger than
23.         zero on entry, decrement it by one and return immediately. If it is zero
24.         on entry, block, waiting until some other thread has called release() to
25.         make it larger than zero. This is done with proper interlocking so that
26.         if multiple acquire() calls are blocked, release() will wake exactly one
27.         of them up. The implementation may pick one at random, so the order in
28.         which blocked threads are awakened should not be relied on. There is no
29.         return value in this case.
31.         When invoked with blocking set to true, do the same thing as when called
32.         without arguments, and return true.
34.         When invoked with blocking set to false, do not block. If a call without
35.         an argument would block, return false immediately; otherwise, do the
36.         same thing as when called without arguments, and return true.
38.         When invoked with a timeout other than None, it will block for at
39.         most timeout seconds.  If acquire does not complete successfully in
40.         that interval, return false.  Return true otherwise.
42.         """
43.         if not blocking and timeout is not None:
44.             raise ValueError("can't specify timeout for non-blocking acquire")
45.         rc = False
46.         endtime = None
47.         with self._cond:
48.             while self._value == 0:
49.                 if not blocking:
50.                     break
51.                 if timeout is not None:
52.                     if endtime is None:
53.                         endtime = _time() + timeout
54.                     else:
55.                         timeout = endtime - _time()
56.                         if timeout <= 0:
57.                             break
58.                 self._cond.wait(timeout)
59.             else:
60.                 self._value -= 1
61.                 rc = True
62.         return rc
64.     __enter__ = acquire
66.     def release(self, n=1):
67.         """Release a semaphore, incrementing the internal counter by one or more.
69.         When the counter is zero on entry and another thread is waiting for it
70.         to become larger than zero again, wake up that thread.
72.         """
73.         if n < 1:
74.             raise ValueError('n must be one or more')
75.         with self._cond:
76.             self._value += n
77.             for i in range(n):
78.                 self._cond.notify()
80.     def __exit__(self, t, v, tb):
81.         self.release()

复制代码

 
由此，最关键的实现完成了，接下来只需要给 _PySimpleQueue.get 方法也增加 priority 参数，并传入 Semaphore.acquire 方法。 Semaphore.acquire 方法增加 priority 参数，并传入给 Condition.wait 方法，就完成啦，限于篇幅这里就不全写下来了。
另外这里固然加入了 priority 参数，但完全不使用这个参数时，其行为和原始版本时没有区别的，即依然符合 FIFO 策略。
 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。