0

我已经问过这个关于从 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 使用它，而其他值似乎不起作用，所以......

4

1 回答 1

1

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

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

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

我希望这有帮助。

于 2014-05-09T07:54:28.583 回答