Python:如何将进程轮询和无阻塞WebSocket服务器结合在一起?

问题描述

我有个主意.编写基于WebSocket的RPC，该RPC将根据以下情况处理消息.

I have an idea. Write a WebSocket based RPC that would process messages according to the scenario below.

1. 客户端连接到WS(Web套接字)服务器
2. 客户端将消息发送到WS服务器
3. WS服务器将消息放入传入队列(可以是multiprocessing.Queue或RabbitMQ队列)
4. 流程池中的一名工作人员捡起消息进行处理
5. 消息正在处理(速度非常快或非常慢-与WS服务器无关)
6. 处理完邮件后，处理结果将推送到待发队列中
7. WS服务器从队列中弹出结果并将其发送给客户端

注意:关键点在于WS服务器应该是非阻塞的，并且仅负责:

NOTE: the key point is that the WS server should be non-blocking and responsible only for:

• 连接接受
• 从客户端获取消息并将其放入传入队列中
• 从待发队列中弹出消息并将其发送回客户端

注意2::最好以某种方式存储客户端标识符，并将其与来自客户端的消息一起传递

NOTE2: it might be a good idea to store client identifier somehow and pass it around with the message from the client

注意3 :完全正常的是，由于来回排队消息，简单消息处理的速度(例如，获取消息作为输入并将其推回)的速度应降低.目标目标是能够以与处理快速消息相同的代码样式在池中运行处理器昂贵的操作(粗略的非实际示例:多个嵌套的"for"循环). IE.从输入队列中弹出消息以及某种客户端标识符，对其进行处理(可能要花一些时间)，然后将处理结果与客户端ID一起推送到输出队列中.

NOTE3: it is completely fine that because of queueing the messages back and forth the speed of simple message processing (e.g. get message as input and push it back as a result) shall become lower. Target goal is to be able to run processor expensive operations (rough non-practical example: several nested "for" loops) in the pool with the same code style as handling fast messages. I.e. pop message from the input queue together with some sort of client identifier, process it (might take a while) and push the processing results together with client ID to the output queue.

问题:

• 在TornadoWeb中，如果我有一个队列(多处理或Rabit)，如何每当有新事件发生时，我都会让Tornado的IOLoop触发一些回调该队列中的项目?你能把我导航到一些现有的吗实施，如​​果有的话?
• 有没有这样的设计现成的实施方案? (不一定是龙卷风)
• 也许我应该使用另一种语言(而不是python)来实现这样的设计?

• In TornadoWeb, if I have a queue (multiprocessing or Rabit), how canI make Tornado’s IOLoop trigger some callback whenever there is a newitem in that queue? Can you navigate me to some existingimplementation if there is any?
• Is there any ready implementation of such a design? (Not necessarily with Tornado)
• Maybe I should use another language (not python) to implement such a design?

致谢:

• 不建议将REST和WSGI用于我要实现的任何目标
• 诸如此处是我通过谷歌搜索2秒钟发现的代码的链接"之类的评论.它有一些来自龙卷风和多处理程序的进口.我不确定它的作用，但是我99％确信它正是您所需要的"，也不受欢迎.
• 建议使用异步库而不是普通的阻塞库... :)

推荐答案

Tornado的IOLoop允许您通过文件描述符处理来自任何文件对象的事件，因此您可以尝试以下操作:

Tornado's IOLoop allows you handling events from any file object by its file descriptor, so you could try this:

• 通过multiprocessing.Pipe
与您的每个工作流程建立联系为每个管道的父级
• 调用 add_handler 结束(使用连接的fileno())
• 使工作人员每次将某些内容放入输出队列时都写一些随机垃圾，无论该值是否为任何MQ的multiprocessing.Queue.
• 在事件处理程序中处理工人的答案

• connect with each of your workers processes through multiprocessing.Pipe
• call add_handler for each pipe's parent end (using the connection's fileno())
• make the workers write some random garbage each time they put something into the output queue, no matter if that's multiprocessing.Queue of any MQ.
• handle the answers form the workers in the event handlers

这篇关于Python:如何将进程轮询和无阻塞WebSocket服务器结合在一起?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！