RabbitMQ，EasyNetQ 与 NodeJS (amqplib) - 订阅

Question

我已经问过这个关于从 amqplib 发布的问题 --> EasyNetQ，并得到了 EasyNetQ 作者的帮助。

现在，我在走另一条路时遇到了麻烦。

它确实短暂地“工作”了，但后来我回去清理了我创建的所有队列，现在 - 它不起作用（从 amqplib 发布到 ENQ 仍然有效，但 ENQ 到 amqplib 没有)。

如果我有这个代码：

Bus.SubscribeAsync<BusManifestHolla>(HollaSubID_1,
    msg => Task.Factory.StartNew(() => {
        Console.WriteLine("LOOK===> Received Manifest Holla ID {0}", msg.ManifestID.ToString());
        Console.WriteLine("LOOK===> Responding with Manifest Yo ID {0}", HollaSubID_1);
        Bus.PublishAsync(new BusManifestYo { ManifestID = msg.ManifestID, ServiceName = HollaSubID_1 });
    })
);

我需要在下面的Node/amqplib中插件什么来订阅/消费它？

Play.AMPQ.then((connection) => {
    return connection.createChannel().then((channel) => {
        return channel.assertExchange(dto.BusManifestYo.Type, 'topic', { durable: true, autoDelete: false }).then((okExchangeReply) => {
            return channel.assertQueue(dto.BusManifestYo.Type).then((ok) => {
                return channel.consume(ok.queue, (msg) => {
                    console.log(util.format('Received message: %s', msg.content.toString()));
                    var bmy: dto.interfaces.IBusManifestYo = JSON.parse(msg.content.toString());

                    channel.ack(msg);
                });
            });
        });
    });
});

更新

如果我有 EasyNetQ 首先创建队列（它将发布到），然后删除节点中的 'assertQueue' 调用（这样它不会破坏队列），然后遵循命名约定 - 它可以工作。当然，这不是一个真正的解决方案，但也许它会帮助某人指出解决方案？

更新 #2

好吧，显然我需要将队列绑定到交换机。这是新的工作代码：

Play.AMPQ.then((connection) => {
    return connection.createChannel().then((channel) => {
        channel.on('error', Play.handleChannelError);

        return channel.assertQueue(dto.BusManifestYo.Type + '_Node', { durable: true, exclusive: false, autoDelete: false }).then((okQueueReply) => {
            return channel.assertExchange(dto.BusManifestYo.Type, 'topic', { durable: true, autoDelete: false }).then((okExchangeReply) => {
                return channel.bindQueue(dto.BusManifestYo.Type + '_Node', dto.BusManifestYo.Type, '#').then((okBindReply) => {
                    return channel.consume(dto.BusManifestYo.Type + '_Node', (msg) => {
                        console.log(util.format('Received message: %s', msg.content.toString()));
                        var bmy: dto.interfaces.IBusManifestYo = JSON.parse(msg.content.toString());

                        channel.ack(msg);
                    });
                });
            });
        });
    });
});

我不清楚的一件事是我将绑定上的模式设置为“#”。它正在工作，但我之所以这么说只是因为我看到 ENQ 使用它，而其他值似乎不起作用......

Answer 1

EasyNetQ 有它自己的交换绑定队列约定。当它订阅时，它会按照您的发现：

1. 创建以消息类型命名的主题交换。
2. 创建一个以消息类型和订阅 ID 命名的队列。
3. 使用“#”绑定键绑定它们。

它使用主题交换（而不是直接）的原因是为了支持主题路由，这就是我们使用“#”（给我所有）绑定的原因。如果您不使用主题发布，则消息会使用空白路由键发布，并且只会路由到“#”绑定。

我希望这会有所帮助。