python - 用于寻址异步请求的 ZMQ Python 请求回复代理

Question

我想利用 ZMQ 来实现(在 python 中)一个代理和一个客户端，用于异步处理对寻址实体的请求-回复。客户端包含执行请求和回复的功能(唯一缺少的是确切的套接字类型/模式)。

请求可以是阻塞的，但回复端需要能够处理传入的并行(线程)请求。(即 REP 套接字不够好，因为它需要在下一次接收之前发送)

它需要通过代理，因为有许多可能的实体可以执行请求和回复，而我只想绑定(bind)一组端口(不是每个实体一个)。

Entity1                Broker                    Entity2
  REQ ------------- ROUTER ?????? -------------- ??????

Entity1 将知道 Entity2 的 ID，并使用它来确保请求是专门向 Entity2 发出的。可以有任意数量的实体，但所有应响应请求的实体都将注册 ID。

我尝试过在上面的代理右侧使用 DEALER，但该经销商似乎只会循环发送请求。

那么有人知道我可以使用一个好的模式/套接字集来异步寻址特定实体吗？

摘要:

• 请求端阻塞
• 用于绑定(bind)固定号码的经纪人/代理端口数量
• 回复套接字应由请求者
• 线程式回复(回复端可以接收和处理并行请求)

我已经相当广泛地阅读了 ZMQ 手册，但我还没有找到任何真正好的模式来通过代理寻址特定套接字，因此非常感谢任何帮助。

Answer 1

经过进一步的研究和测试，我发现了一种模式，它似乎可以为我的所有需求提供解决方案。

模式

Requester               Broker                   Replier
  REQ ------------- ROUTER ROUTER -------------- DEALER
                (requests) (replies)

请求者

客户端的请求端只需连接到代理上的请求路由器，发送请求并开始读取套接字以获取回复:

reqSocket.connect(self._reqAddress)
reqSocket.send_multipart([repId, message])
reply = reqSocket.recv_multipart()[0]

回复者 ID 作为消息的第一部分包含在内，例如:

Outgoing message: ['replierId', 'requestMsg']

请求路由器

if self.reqRouterSocket in socketEvents:
    multipart = self.reqRouterSocket.recv_multipart()
    multipart = [multipart[-2]] + multipart
    del multipart[-2]
    self.repRouterSocket.send_multipart(multipart)

即，请求路由器仅移动有效负载的第一部分(即replierId)并将其放在地址堆栈中的第一个部分:

Incoming message: ['reqSocketAddr', '', 'replierId', 'requestMsg']
Outgoing message: ['replierId', 'reqSocketAddr', '', 'requestMsg']

传出消息是从回复路由器发送的。由于回复者将其套接字 ID 设置为“replierId”并已连接到回复者路由器，因此该路由器可以识别该地址并能够成功传送请求。

回复者

回复者需要将其自己的套接字标识设置为某个已知值，以便如上所述直接寻址。

注意:在执行与回复路由器的连接之前，您必须设置 DEALER 套接字的套接字 ID。设置套接字的身份:

self.dealerSocket.setsockopt(zmq.IDENTITY, 'replierId')

否则路由器将不知道ID并会抛出消息。

应答器监听传入的请求。就我而言，这都是线程化的，并且请求是异步处理的。这就是使用 DEALER 套接字而不是常规 REP 的原因，在同步情况下这会容易得多。 DEALER 套接字可以接收更多请求，而不必首先回答第一个请求，而 REP 必须这样做。然而，回复者端所做的事情的简化版本是:

multipart = self.dealerSocket.recv_multipart()
returnRoute = multipart[:-1]
requestMsg = multipart[-1]
reply = someFunction(requestMsg)
self.dealerSocket.send_multipart(returnRoute + [reply])

即，回复者只返回它得到的内容，但请求已更改为回复:

Incoming message: ['replierId', 'reqSocketAddr', '', 'request']
Outgoing message: ['replierId', 'reqSocketAddr', '', 'reply']

此传出消息随后被发送回回复路由器。

回复路由器

在代理的这一侧选择路由器纯粹是因为它需要在许多连接的套接字中寻址特定套接字的功能。

if self.repRouterSocket in socketEvents:
    multipart = self.repRouterSocket.recv_multipart()
    self.reqRouterSocket.send_multipart(multipart[1:])

即，弹出地址栈的首地址，然后再次将消息发送给请求方。

Incoming message: ['replierId', 'reqSocketAddr', '', 'reply']
Outgoing message: ['reqSocketAddr', '', 'reply']

请求路由器识别该地址并将请求发送回请求者，请求者接收:

Incoming list: ['reply']

这种模式似乎满足我在问题中提出的要求。我希望它对其他人也有用。