gevent 队列因 LoopExit 而失败

Question

我想使用python gevent库来实现一个生产者和多个消费者服务器。这是我的尝试：

class EmailValidationServer():
   def __init__(self):
      self.queue = Queue()
   def worker(self):
      while True:
          json = self.queue.get()
   def handler(self,socket,address):
      fileobj = socket.makefile()
      content = fileobj.read(max_read)
      contents = json.loads(content)
      for content in contents:
          self.queue.put(content)
   def daemon(self,addr='127.0.0.1',num_thread=5):
      pool = Pool(1000)
      server = StreamServer((addr, 6000),self.handler,spawn=pool) # run
      pool = ThreadPool(num_thread)
      for _ in range(num_thread):
          pool.spawn(self.worker)
      server.serve_forever()
if __name__ == "__main__":
    email_server = EmailValidationServer()
    email_server.daemon()

我使用了来自 gevent.queue.Queue 的队列。它给了我错误信息：

LoopExit: This operation would block forever
(<ThreadPool at 0x7f08c80eef50 0/4/5>,
 <bound method EmailValidationServer.worker of <__main__.EmailValidationServer instance at 0x7f08c8dcd998>>) failed with LoopExit

问题：但是当我将队列从 gevent 的实现更改为 python 内置库时，它可以工作。我不知道原因，我想它们的实现之间存在差异是支持的。我不知道 gevent 不允许无限等待的原因。有没有人可以给出解释？提前感谢

Answer 1

我建议您可以使用gevent.queue.JoinableQueue() 代替Python 内置的Queue()。 API使用可以参考官方队列指南(http://www.gevent.org/gevent.queue.html)

def worker():
    while True:
        item = q.get()
        try:
            do_work(item)
        finally:
            q.task_done()

q = JoinableQueue()
for i in range(num_worker_threads):
     gevent.spawn(worker)

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

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

如果你再次遇到异常，你最好完全理解Gevent corouinte控制流的原理......一旦你明白了，那没什么大不了的。 :)