Python:如何结合进程轮询和非阻塞 WebSocket 服务器？

Question

我有一个想法。编写一个基于 WebSocket 的 RPC，它将根据以下场景处理消息。

1. 客户端连接到 WS(网络套接字)服务器
2. 客户端向WS服务器发送消息
3. WS 服务器将消息放入传入队列(可以是 multiprocessing.Queue 或 RabbitMQ 队列)
4. 进程池中的一个worker拿起消息进行处理
5. 正在处理消息(可能非常快或非常慢 - 这与 WS 服务器无关)
6. 消息处理完成后，将处理结果推送到输出队列
7. WS服务器从队列中弹出结果并发送给客户端

注意:关键是WS服务器应该是非阻塞的并且只负责:

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

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

注意 3:完全没问题，因为消息来回排队，简单消息处理的速度(例如，获取消息作为输入并将其推回)会变慢。 Target 目标是能够在池中使用与处理快速消息相同的代码风格来运行处理器昂贵的操作(粗略的非实际示例:几个嵌套的“for”循环)。 IE。从输入队列中弹出消息和某种客户端标识符，对其进行处理(可能需要一段时间)并将处理结果与客户端 ID 一起推送到输出队列。

问题:

• 在 TornadoWeb 中，如果我有一个队列(多处理或 Rabit)，如何只要有新的，我就让 Tornado 的 IOLoop 触发一些回调该队列中的项目？你能把我导航到一些现有的吗有没有实现？
• 是否有现成的此类设计实现方案？ (不一定是 Tornado )
• 也许我应该使用另一种语言(不是 python)来实现这样的设计？

致谢:

• 不欢迎使用 REST 和 WSGI 来实现我想要实现的任何目标
• 评论如“这是我通过谷歌搜索 2 秒找到的代码链接。它从 tornado 和 multiprocessing 中导入了一些内容。我不确定它的作用，但是我 99% 确定它正是您所需要的”也不受欢迎
• 使用异步库而不是普通阻塞库的建议是……:)

Answer 1

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

• 通过 multiprocessing.Pipe 连接您的每个工作进程
• 调用add_handler对于每个管道的父端(使用连接的 fileno())
• 让工作人员在每次将内容放入输出队列时写入一些随机垃圾，无论那是不是任何 MQ 的 multiprocessing.Queue。
• 在事件处理程序中处理来自工作人员的答案