同步来自多个 ZeroMQ 发布者的消息答案

【问题标题】：Synchronising messages from multiple ZeroMQ publishers同步来自多个 ZeroMQ 发布者的消息
【发布时间】：2022-02-06 21:02:03
【问题描述】：

我正在尝试创建一个库存监控应用程序。我有两个 ZeroMQ 发布者正在运行，它们从两个各自的服务中获取当前股票价格。第一个服务是主要的信息来源，不能丢弃来自该服务的任何消息。

我想把两个价格合并成一个对象，然后推送到前端。每条消息都有一个时间戳，表示观察价格的时间，是组合消息的基础。

但是，两条消息不会同时到达，系统必须等待两条消息都到达，然后将它们组合起来，然后将它们推送到前端。

我已经能够使用内存映射来使其工作，该映射存储按时间戳索引的消息并检查两条消息是否都已到达。如果他们没有，它会在发送消息之前等待一段时间。消息传递应该是有序的，并且不能丢弃来自第一个服务的数据。

const messageQueue = new Map();

monitoringServiceOne.on('message', function (topic, message) {
   synchroniseMessages(topic, message);
});

monitoringServiceTwo.on('message', function (topic, message) {
   synchroniseMessages(topic, message);
});

function synchroniseMessages(topicName, incomingMessage) {

    if (topicName == "ServiceOne") {
        const message = messageQueue.get(message.time);

        // If message exists, it means service two message has been received
        if (message) {
            pushToFE({ ... { message }, "ServiceOne": incomingMessage })
        } else {
            setTimeout(() => {
                const message = messageQueue.get(message.time);
                if (message) {
                    pushToFE({ ... { message }, "ServiceOne": incomingMessage })
                }
            }, DELAY)
        }
        messageQueue.set(message.time, { "ServiceOne": incomingMessage });
    } else {
        const message = messageQueue.get(message.time);

        // If message exists, it means service one message has been received
        if (message) {
            pushToFE({ ... { message }, "ServiceTwo": incomingMessage })
        } else {
            messageQueue.set(message.time, { "ServiceTwo": incomingMessage });
        }
    }

}

即使这行得通，我还是想重新发明轮子。是否有任何现有的 ZeroMQ 范例或模式可用于改进代码？我一直在寻找一种事件驱动模式，它允许我定义接收器的逻辑。

【问题讨论】：

标签： node.js message-queue zeromq

【解决方案1】：

简短的回答是否定的，ZMQ 中没有任何东西可以为您做到这一点。这是一个演员模型系统，它所做的只是保证整个消息将按照它们被发送到机器的顺序出现。没有时间保证。

例如，如果有一个 PUB / SUB 设置并且有两个或多个 PUBlisher（我认为是允许的组合），则消息由不同 PUBlishers 发布的时间顺序并不一定会导致它们以该顺序到达在订阅者中。但它确实保证了这些 PUBlisher 中的任何一个发送的多条消息都不会在传输过程中被交换。

因此，如果您的 ServiceOne 和 ServiceTwo 是同一个 ZMQ 套接字上的 PUBlishers，那么您就无法确定按 1a、2a、1b、2b、1c、2c、1d、2d、1e、2e、1f 的顺序获取消息, 2f, 1g, 2g。它们可能以 1a、2a、1b、1c、2b、2c、2d、1d、2e、1e、1f、2f、1g、2g 的形式到达。数字不必按顺序排列，但任何给定数字的字母都可以。

我喜欢使用地图（字典？我的 node.js 知识非常有限......）的想法，优雅。据推测，您必须时不时地修剪地图，例如如果 ServiceOne 断开连接一段时间，地图可能会变得非常大。

呃，在调用messageQueue.get() 时应该使用incomingMessage.time，所以messageQueue.get(incomingMessage.time) 和messageQueue.set() 类似？

【讨论】：