如何在RabbitMQ中使用单个C#消费者订阅多个流？

可以用单个消费者订阅多个RabbitMQ流，不需要创建多个独立的消费者实例，基于你现有的代码稍作改造就能实现。

核心改造点 #

RabbitMQ Stream Client的ConsumerConfig支持通过Streams属性指定多个要订阅的流，同时可以在消息处理函数中通过stream参数区分消息来自哪个流。

改造后的完整代码 #

using RabbitMQ.Stream.Client;
using RabbitMQ.Stream.Client.Reliable;
using System.Net;
using System.Net.Sockets;

var streamSystem = await StreamSystem.Create(
    new StreamSystemConfig()
    {
        UserName = "guest",
        Password = "guest",
        Endpoints = new List<EndPoint>
        {
            new IPEndPoint(IPAddress.Parse("127.0.0.1"), 5552)
        }
    }
).ConfigureAwait(false);

var confirmationTaskCompletionSource = new TaskCompletionSource<int>();
var consumer = await Consumer.Create(
    new ConsumerConfig(streamSystem) // 这里不再传单个流名
    {
        // 指定要订阅的多个流
        Streams = new List<string> { "my-stream-1", "my-stream-2", "my-stream-3" },
        // 全局偏移配置（所有流都用相同起始偏移）
        OffsetSpec = new OffsetTypeFirst(),
        MessageHandler = async (stream, consumer, context, message) =>
        {
            // 通过stream参数区分消息来源
            Console.WriteLine($"从流 [{stream}] 收到消息");
            await Task.CompletedTask.ConfigureAwait(false);
        }
    }
).ConfigureAwait(false);

// 保持消费者运行，按需添加阻塞逻辑
await confirmationTaskCompletionSource.Task.ConfigureAwait(false);

进阶：为每个流单独配置偏移 #

如果不同流需要不同的起始偏移，可以使用StreamOffsetSpecs属性代替全局的OffsetSpec：

StreamOffsetSpecs = new Dictionary<string, OffsetSpec>
{
    { "my-stream-1", new OffsetTypeFirst() },
    { "my-stream-2", new OffsetTypeLast() },
    { "my-stream-3", new OffsetTypeOffset(100) } // 从偏移量100开始消费
}

注意事项 #

• 确保使用的RabbitMQ Stream Client版本≥1.0，旧版本可能不支持多流订阅
• 单个消费者会并行处理来自不同流的消息，但同一个流内的消息依然保持顺序投递
• 消费性能取决于流的消息量和消费者的处理能力，如果流的消息吞吐量极高，可考虑拆分多个消费者

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