Python Gevent共享队列(监听进程)

Question

我正在尝试让一些代码工作，我可以使用 gevent 实现登录到多线程程序。我想要做的是设置自定义日志记录处理程序以将日志事件放入队列中，同时监听器进程不断监视新的日志事件以进行适当处理。我过去曾使用 Multiprocessing 完成过此操作，但从未使用过 Gevent。

我遇到一个问题，程序陷入无限循环(监听器进程)，并且不允许其他线程“工作”...

理想情况下，工作进程完成后，我可以将任意值传递给监听器进程以告诉它打破循环，然后将所有进程连接在一起。这是我到目前为止所拥有的:

import gevent
from gevent.pool import Pool
import Queue
import random
import time

def listener(q):
    while True:
        if not q.empty():
            num = q.get()
            print "The number is: %s" % num
            if num <= 100:
                print q.get()
            # got passed 101, break out
            else:
                break
        else:
            continue
def worker(pid,q):
    if pid == 0:
        listener(q)
    else:
        gevent.sleep(random.randint(0,2)*0.001)
        num = random.randint(1,100)
        q.put(num)

def main():
    q = Queue.Queue()
    all_threads = []
    all_threads = [gevent.spawn(worker, pid,q) for pid in xrange(10)]
    gevent.wait(all_threads[1:])
    q.put(101)
    gevent.joinall(all_threads)

if __name__ == '__main__':
    main()

正如我所说，该程序似乎卡在第一个进程上，不允许其他工作人员做他们的事情。我还尝试过完全独立地生成监听器进程(这实际上是我宁愿这样做的方式)，但这似乎也不起作用，所以我尝试了这种方式。

任何帮助将不胜感激，感觉我可能只是错过了一些关于 gevent 后端的明显内容。

谢谢

Answer 1

第一个问题是，如果队列最初为空，您的监听器永远不会屈服。您生成的第一个任务是您的监听器。当它开始时，有一个 while True:，q 将为空，因此您转到 else 分支，该分支继续，循环回到 while 循环的开头，然后 q 是仍然是空的。所以你只需坐在第一个线程中不断检查 q 是否为空。

这里的关键是 gevent 不使用“ native ”线程或进程。与“真实”线程不同，“真实”线程可以通过幕后的某些东西(例如操作系统调度程序)随时切换到，gevent 使用“greenlet”，它要求您执行某些操作以“将控制权交给”另一个任务。该内容是 gevent 认为会阻塞的任何内容，例如从网络、磁盘读取或使用阻塞 gevent 操作之一。

一个粗略的修复方法是在 pid == 9 而不是 0 时启动监听器。通过使其最后生成，q 中将会有项目，并且它将进入主 if分支。缺点是这并不能解决逻辑问题，因此第一次队列为空时，您将再次陷入无限循环。

更正确的修复方法是使用 gevent.sleep() 而不是 continue。 sleep 是一个阻塞操作，因此您的其他任务将有机会运行。如果没有参数，它不会等待任何时间，但如果 gevent 准备好运行，它仍然有机会决定切换到另一个任务。但这仍然不是很有效，就好像队列是空的一样，它将花费大量无意义的时间一遍又一遍地检查并要求尽快再次运行。 sleep 时间比默认值 0 更长会更有效，但会延迟处理日志消息。

但是，您可以利用这样一个事实:许多 gevent 类型(例如 Queue)可以以更 Python 的方式使用，并使您的代码更简单、更容易理解，也更高效。

import gevent
from gevent.queue import Queue

def listener(q):
    for msg in q:
        print "the number is %d" % msg

def worker(pid,q):
    gevent.sleep(random.randint(0,2)*0.001)
    num = random.randint(1,100)
    q.put(num)

def main():
    q = Queue()
    listener_task = gevent.spawn(listener, q)
    worker_tasks = [gevent.spawn(worker, pid, q) for pid in xrange(1, 10)]
    gevent.wait(worker_tasks)
    q.put(StopIteration)
    gevent.join(listener_task)

这里，Queue 可以作为 for 循环中的迭代器进行操作。只要有消息，它就会获取一个项目，运行循环，然后等待另一个项目。如果没有元素，它就会阻塞并徘徊，直到下一个元素到达。不过，由于它会阻塞，gevent 将切换到您的其他任务之一来运行，从而避免示例代码存在的无限循环问题。

因为这个版本使用Queue作为for循环迭代器，所以我们还可以自动将一个很好的哨兵值放入队列中以使监听器任务退出。如果 for 循环从其迭代器中获取 StopIteration，它将干净地退出。因此，当从 q 读取的 for 循环从 q 获取 StopIteration 时，它退出，然后函数退出，并且生成的任务完成。