我在 python 中使用队列，它的工作方式类似于 channel 。也就是说，无论何时进行插入，其他线程都在等待并获取插入的值。该值就是 yield 。
@classsynchronized('mutex')
def send(self, message):
# This might raise Full
if not self._is_closed:
self._queue.put(message)
return True
return False
@classsynchronized('mutex')
def close(self):
# Don't do what we don't need to
if self._is_closed:
return
# Make the _queue.get() request fail with an Empty exception
# This will cause the channel to stop listenning to messages
# First aquire the write lock, then notify the read lock and
# finally release the write lock. This is equivalent to an
# empty write, which will cause the Empty exception
print("ACQUIRING not_full")
self._queue.not_full.acquire()
# Close first. If the queue is empty it will raise Empty as fast as
# possible, instead of waiting for the timeout
self._is_closed = True
try:
print("NOTIFYING not_empty")
self._queue.not_empty.notify()
print("NOTIFIED not_empty")
finally:
self._queue.not_full.release()
print("RELEASED not_full")
def _yield_response(self):
try:
while True:
# Fetch from the queue, wait until available, or a timeout
timeout = self.get_timeout()
print("[WAITING]")
message = self._queue.get(True, timeout)
print("[DONE WAITING] " + message)
self._queue.task_done()
# Don't yield messages on closed queues, flush instead
# This prevents writting to a closed stream, but it's
# up to the user to close the queue before closing the stream
if not self._is_closed:
yield message
# The queue is closed, ignore all remaining messages
# Allow subclasses to change the way ignored messages are handled
else:
self.handle_ignored(message)
# This exception will be thrown when the channel is closed or
# when it times out. Close the channel, in case a timeout caused
# an exception
except Empty:
pass
# Make sure the channel is closed, we can get here by timeout
self.close()
# Finally, empty the queue ignoring all remaining messages
try:
while True:
message = self._queue.get_nowait()
self.handle_ignored(message)
except Empty:
pass
我只包含了相关的方法，但请注意这是一个类。问题是，这并不像我预期的那样。队列确实关闭了，所有打印都显示在控制台中，但是等待消息的线程没有得到通知。相反，它总是超时退出。
所有@classsynchronized('mutex') 注解在类方面同步具有相同标识符 ('mutex') 的方法，也就是说，具有相同 ID 的注解的类中的每个方法都相互同步。
我在关闭前获取not_full锁的原因是为了防止在关闭的 channel 中插入。只有这样我才会通知 not_empty 锁。
知道为什么这行不通吗？还有其他建议吗？
提前致谢。
编辑:
我对打印品做了一些改动。我创建 channel 并立即发送消息。然后我发送一个删除它的 HTTP 请求。这是输出:
[WAITING]
[DONE WAITING] message
[WAITING]
ACQUIRING not_full
NOTIFYING not_empty
NOTIFIED not_empty
RELEASE not_full
所以:
- 第一条消息得到处理并成功发送(我在客户端获取它，所以...)
- 然后队列正在等待。它应该在等待 not_empty 锁，对吧？
- 我对 channel 发出删除请求。它获取 not_full 锁(以防止写入)，并通知 not_empty 锁。
我真的不明白...如果线程收到通知为什么它不解除阻塞？？
请您参考如下方法:
篡改 Queue 的内部锁似乎是个坏主意。只用Queue的官方接口(interface)来表述问题如何？
例如，要模拟关闭队列，您可以执行 self._queue.put(None) 或其他一些特殊值。得到这个特殊值的等待线程知道队列已经关闭。问题是特殊值不再在队列中以供更多线程查看；但这很容易解决:当线程获得特殊值时，它会立即将其再次放入队列中。
