ZeroMQ 重传 pub-sub 模型答案

【问题标题】：ZeroMQ retransmitting pub-sub modelZeroMQ 重传 pub-sub 模型
【发布时间】：2020-10-13 02:37:19
【问题描述】：

我正在尝试使用 ZeroMQ javascript 绑定来实现重新传输的 pub-sub 主干。

大意是：

入站模块向主干宣布自己，主干订阅它们并在自己的发布者套接字上重新传输
出站模块订阅骨干网

我遇到了一个需要从单个线程使用的 pub 套接字的问题。

我当前的代码是这样的：

async function listen(name: string, sub: zmq.Subscriber, pub: zmq.Publisher) {
    let id = 0;
    for await (const [topic, msg] of sub) {
        console.log(`BACKBONE | ${name} | received a message id: ${++id} related to: ${topic.toString()} containing message: ${msg.toString()}`);
        await pub.send([topic, msg]);
    }
}

每个入站模块都会实例化，但当然它们会在 pub.send 上发生冲突。

【问题讨论】：

标签： node.js typescript zeromq

【解决方案1】：

我写了一个队列来序列化socket访问，解决了这个问题。

import * as zmq from "zeromq"

export class PublisherQueue {
    private queue: Buffer[][] = []
    constructor(private publisher: zmq.Publisher) { }

    private static toBuffer(value: Buffer | string): Buffer {
        if (value instanceof Buffer) {
            return value as Buffer;
        } else {
            return Buffer.from(value as string);
        }
    }

    send(topic: Buffer | string, msg: Buffer | string) {
        this.queue.push([PublisherQueue.toBuffer(topic), PublisherQueue.toBuffer(msg)]);
    }

    async run() {
        while (true) {
            if (this.queue.length > 0) {
                let msg = this.queue.shift();
                if (msg !== undefined) {
                    await this.publisher.send(msg);
                }
            } else {
                await new Promise(resolve => setTimeout(resolve, 50));
            }
        }
    }
}

【讨论】：