我已经实现了一个消费者/生产者优先级队列，其中优先级实际上是一个时间戳，表示何时应该交付项目。它工作得很好，但我想知道是否有人有更好的想法来实现这个或对当前实现的评论。
代码是用 Python 编写的。创建单个线程以按时唤醒等待的消费者。我知道这是在库中创建线程的反模式，但我无法设计其他方法。
代码如下:
import collections
import heapq
import threading
import time
class TimelyQueue(threading.Thread):
"""
Implements a similar but stripped down interface of Queue which
delivers items on time only.
"""
class Locker:
def __init__(self, lock):
self.l = lock
def __enter__(self):
self.l.acquire()
return self.l
def __exit__(self, type, value, traceback):
self.l.release()
# Optimization to avoid wasting CPU cycles when something
# is about to happen in less than 5 ms.
_RESOLUTION = 0.005
def __init__(self):
threading.Thread.__init__(self)
self.daemon = True
self.queue = []
self.triggered = collections.deque()
self.putcond = threading.Condition()
self.getcond = threading.Condition()
# Optimization to avoid waking the thread uselessly.
self.putwaketime = 0
def put(self, when, item):
with self.Locker(self.putcond):
heapq.heappush(self.queue, (when, item))
if when < self.putwaketime or self.putwaketime == 0:
self.putcond.notify()
def get(self, timeout=None):
with self.Locker(self.getcond):
if len(self.triggered) > 0:
when, item = self.triggered.popleft()
return item
self.getcond.wait(timeout)
try:
when, item = self.triggered.popleft()
except IndexError:
return None
return item
def qsize(self):
with self.Locker(self.putcond):
return len(self.queue)
def run(self):
with self.Locker(self.putcond):
maxwait = None
while True:
curtime = time.time()
try:
when, item = self.queue[0]
maxwait = when - curtime
self.putwaketime = when
except IndexError:
maxwait = None
self.putwaketime = 0
self.putcond.wait(maxwait)
curtime = time.time()
while True:
# Don't dequeue now, we are not sure to use it yet.
try:
when, item = self.queue[0]
except IndexError:
break
if when > curtime + self._RESOLUTION:
break
self.triggered.append(heapq.heappop(self.queue))
if len(self.triggered) > 0:
with self.Locker(self.getcond):
self.getcond.notify()
if __name__ == "__main__":
q = TimelyQueue()
q.start()
N = 50000
t0 = time.time()
for i in range(N):
q.put(time.time() + 2, i)
dt = time.time() - t0
print "put done in %.3fs (%.2f put/sec)" % (dt, N / dt)
t0 = time.time()
i = 0
while i < N:
a = q.get(3)
if i == 0:
dt = time.time() - t0
print "start get after %.3fs" % dt
t0 = time.time()
i += 1
dt = time.time() - t0
print "get done in %.3fs (%.2f get/sec)" % (dt, N / dt)
请您参考如下方法:
这里唯一真正需要后台线程的是一个定时器，当它用完时踢服务员，对吧？
首先，您可以使用 threading.Timer 来实现它而不是显式的后台线程。但是，虽然这可能更简单，但它并不能真正解决您在用户背后创建线程的问题，无论他们是否愿意。此外，对于 threading.Timer，每次重新启动计时器时实际上都在分离一个新线程，这可能是一个性能问题。 (您一次只有一个，但启动和停止线程仍然不是免费的。)
如果您环顾 PyPI 模块、ActiveState 配方和各种框架，有许多实现可以让您在单个后台线程上运行多个计时器。那将解决您的问题。
但这仍然不是一个完美的解决方案。例如，假设我的应用程序需要 20 个 TimelyQueue 对象——或者一个 TimelyQueue 加上 19 个其他都需要计时器的对象。我仍然会得到 20 个线程。或者，假设我正在构建一个套接字服务器或一个 GUI 应用程序(TimelyQueue 的两个最明显的用例；我可以在我的事件循环之上实现一个计时器(或者，最有可能的是，只需使用框架附带的计时器)，那么我为什么还需要线程呢？
解决这个问题的方法是提供一个钩子(Hook)来提供任何定时器工厂:
def __init__(self, timerfactory = threading.Timer):
self.timerfactory = timerfactory
...
现在，当您需要调整计时器时:
if when < self.waketime:
self.timer.cancel()
self.timer = self.timerfactory(when - now(), self.timercallback)
self.waketime = when
对于快速和肮脏的用例，开箱即用就足够了。但是如果我，例如，使用 twisted ，我可以只使用 TimelyQueue(twisted.reactor.callLater)，现在队列的计时器通过 twisted 事件循环。或者，如果我有一个多计时器单线程实现，我在别处使用 TimelyQueue(multiTimer.add)，现在队列的计时器与我的所有计时器都在同一个线程上其他计时器。
如果您愿意，您可以提供比 threading.Timer 更好的默认值，但实际上，我认为大多数需要比 threading.Timer 更好的东西的人将是能够为他们的特定应用提供比您提供的任何东西更好的东西。
当然，并非每个计时器实现都具有与 threading.Timer 相同的 API——尽管您会惊讶于它们中有多少是这样做的。但是编写一个适配器并不难，如果你有一个定时器，你想与 TimelyQueue 一起使用，但它有错误的接口(interface)。例如，如果我正在构建 PyQt4/PySide 应用程序，QTimer没有 cancel 方法，需要毫秒而不是秒，所以我必须做这样的事情:
class AdaptedQTimer(object):
def __init__(self, timeout, callback):
self.timer = QTimer.singleShot(timeout * 1000, callback)
def cancel(self):
self.timer.stop()
q = TimelyQueue(AdaptedQTimer)
或者，如果我想更直接地将队列集成到 QObject 中，我可以包装 QObject.startTimer() 并让我的 timerEvent(self ) 方法调用回调。
一旦您考虑适配器，最后一个想法。我认为这不值得，但可能值得考虑。如果您的计时器采用时间戳而不是时间增量，并且具有 adjust 方法而不是/代替 cancel，并且拥有自己的 waketime ，您的 TimelyQueue 实现可能更简单，也可能更高效。在 put 中，你有这样的东西:
if self.timer is None:
self.timer = self.timerfactory(when)
elif when < self.timer.waketime:
self.timer.adjust(when)
当然大部分定时器不提供这个接口(interface)。但是，如果有人拥有一个，或者愿意制作一个，他们就能从中受益。对于其他人，您可以提供一个简单的适配器，将 threading.Timer 样式的计时器转换为您需要的类型，例如:
def timerFactoryAdapter(threadingStyleTimerFactory):
class TimerFactory(object):
def __init__(self, timestamp, callback):
self.timer = threadingStyleTimerFactory(timestamp - now(), callback)
self.callback = callback
def cancel(self):
return self.timer.cancel()
def adjust(self, timestamp):
self.timer.cancel()
self.timer = threadingStyleTimerFactory(timestamp - now(), self.callback)
