RabbitMQ sharded queues场景下单个消费者能否消费交换器下全部分片队列

核心结论 #

可以用单个消费者消费同一交换机下的所有分片队列，你当前的实现有两个核心错误导致只能收到部分消息：

• 混淆了交换机和队列的API：QueueDeclarePassive用于队列被动校验，校验交换机应该调用ExchangeDeclarePassive，你这行代码实际是在校验名为TestExchange的队列是否存在，和你要操作的交换机完全无关
• BasicConsume的queue参数必须传入实际的队列名，你传入交换机名的情况下，实际只会消费名为TestExchange的单个队列，自然拿不到其他分片队列的消息

分片队列消费逻辑说明 #

RabbitMQ分片队列插件会自动在集群节点上生成多个带分片后缀的实际队列，命名规则一般为{交换机名}-{分片序号}，比如你的TestExchange对应分片队列就是TestExchange-0、TestExchange-1等。你需要先拿到该交换机对应的所有分片队列名称，再将同一个消费者实例绑定到所有这些队列上即可。

修改后代码示例 #

using (var connection = factory.CreateConnection())
{
    using (var channel = connection.CreateModel())
    {
        // 先校验目标交换机存在
        channel.ExchangeDeclarePassive("TestExchange");
        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += (model, ea) =>
        {
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            Console.WriteLine(" [x] Received {0} from queue: {1}", message, ea.RoutingKey);
        };
        // 这里替换为你实际的分片队列列表，也可通过RabbitMQ管理接口查询该交换机下绑定的所有分片队列
        var shardQueues = new List<string> {"TestExchange-0", "TestExchange-1", "TestExchange-2"};
        foreach (var queueName in shardQueues)
        {
            // 同一个消费者绑定所有分片队列
            channel.BasicConsume(queue: queueName,
                                 autoAck: true,
                                 consumer: consumer);
        }
        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}

补充说明 #

如果你不知道分片队列的具体数量和名称，可以通过RabbitMQ的本地管理接口查询指定交换机下绑定的所有队列列表，再动态绑定即可。

内容的提问来源于stack exchange，提问作者Josh