python - 具有阻塞请求的 Gevent 异步服务器

Question

我认为 Gevent 有一个非常常见的用例。我需要一个监听请求的 UDP 服务器，并根据请求将 POST 提交到外部 Web 服务。外部 Web 服务基本上一次只允许一个请求。

我想要一个异步 UDP 服务器，以便可以立即检索和存储数据，这样我就不会错过任何请求(这部分很容易使用 DatagramServer gevent 提供)。然后我需要一些方法来串行向外部 Web 服务发送请求，但不会破坏 UDP 服务器的异步。

我首先尝试使用 monkey patching everything，我最终得到的是一个快速的解决方案，但是我对外部 Web 服务的请求在任何方面都没有速率限制，这导致了错误。

似乎我需要的是一个非阻塞工作线程，在 UDP 服务器将任务添加到非阻塞工作线程正在工作的队列中时，以串行方式向外部 Web 服务发送请求。

我需要的是有关运行带有用于其他任务(尤其是队列)的额外 greenlet 的 gevent 服务器的信息。我一直在使用 DatagramServer 的 serve_forever 函数，并认为我需要使用 start 方法来代替，但没有找到太多关于它如何实现的信息合二为一。

谢谢，

编辑

答案非常有效。我已经改编了 UDP server example使用 @mguijarr 的答案编写代码，为我的用例生成一个工作示例:

from __future__ import print_function
from gevent.server import DatagramServer
import gevent.queue
import gevent.monkey
import urllib

gevent.monkey.patch_all()

n = 0

def process_request(q):
    while True:
        request = q.get()
        print(request)
        print(urllib.urlopen('https://test.com').read())


class EchoServer(DatagramServer):
    __q = gevent.queue.Queue()
    __request_processing_greenlet = gevent.spawn(process_request, __q)

    def handle(self, data, address):
        print('%s: got %r' % (address[0], data))
        global n
        n += 1
        print(n)
        self.__q.put(n)
        self.socket.sendto('Received %s bytes' % len(data), address)


if __name__ == '__main__':
    print('Receiving datagrams on :9000')
    EchoServer(':9000').serve_forever()

Answer 1

下面是我的做法:

1. 编写一个以“队列”对象作为参数的函数；此功能将不断处理队列中的项目。每个项目都应该是对 Web 服务的请求。此函数可以是模块级函数，而不是您的 DatagramServer 实例的一部分:

   def process_requests(q):
       while True:
           request = q.get()
           # do your magic with 'request'
           ...
   

2. 在您的 DatagramServer 中，使函数在 greenlet 中运行(类似于后台任务):

   self.__q = gevent.queue.Queue()
   self.__request_processing_greenlet = gevent.spawn(process_requests, self.__q)
   

3. 当您在 DatagramServer 实例中收到 UDP 请求时，您将请求推送到队列

   self.__q.put(request)    
   

这应该做你想做的。您仍然在 DatagramServer 上调用“serve_forever”，没问题。