单线程RMQ消费服务占用多队列未确认消息问题求助

针对你遇到的单线程Service Shared占用多队列未确认消息的问题，给你三个可行的解决方向：

1. 所有队列共享单个RMQ通道 #

EasyNetQ默认会给每个队列订阅创建独立通道，每个通道的预取计数是独立生效的——这就是单线程能同时拿到多个队列各一条消息的原因。你可以改成让所有队列的订阅共用同一个通道，这样通道级别的预取计数1会全局生效，同一时间只会从任意一个队列获取一条消息，处理完之后才会取下一条。

具体实现示例：

// 创建高级总线实例
var advancedBus = RabbitHutch.CreateBus("host=localhost").Advanced;

// 创建单个通道并设置预取计数
var channel = advancedBus.OpenChannel();
channel.BasicQos(0, 1, false);

// 所有队列订阅都使用这个通道
advancedBus.Consume(queue1, channel, (message, info) => {
    // 处理消息逻辑
    message.Ack();
});

advancedBus.Consume(queue2, channel, (message, info) => {
    // 处理消息逻辑
    message.Ack();
});

2. 改用手动确认+主动拉取模式 #

关闭自动确认机制，改用手动确认，同时将消费模式从默认的推送改成主动拉取。这样你可以严格控制：只有处理完当前消息后，才从任意一个队列拉取下一条消息，从根源上避免同时持有多个未确认消息。

示例代码：

var bus = RabbitHutch.CreateBus("host=localhost");
var advancedBus = bus.Advanced;

var channel = advancedBus.OpenChannel();
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) => {
    var body = ea.Body.ToArray();
    var message = Encoding.UTF8.GetString(body);
    
    // 处理消息逻辑
    Console.WriteLine("处理消息: {0}", message);
    
    // 处理完成后手动确认
    model.BasicAck(ea.DeliveryTag, false);
    
    // 确认后主动拉取下一条
    model.BasicGet(queueName, false);
};

// 初始拉取第一条消息
channel.BasicGet(queueName, false);

3. 给Service Shared加全局单线程任务队列 #

在Service Shared内部维护一个单线程的任务队列，所有队列的消费回调仅把消息丢进这个任务队列，不直接处理。同时用信号量控制：任务队列有未完成任务时，暂停接收新消息；任务完成后再恢复接收。

示例代码：

// 单线程任务队列
var taskQueue = new BlockingCollection<object>();
var semaphore = new SemaphoreSlim(1, 1);

// 队列1的消费者
bus.Subscribe<MessageType1>("queue1_sub", msg => {
    semaphore.Wait();
    taskQueue.Add(msg);
    semaphore.Release();
});

// 队列2的消费者
bus.Subscribe<MessageType2>("queue2_sub", msg => {
    semaphore.Wait();
    taskQueue.Add(msg);
    semaphore.Release();
});

// 单线程处理任务队列中的消息
Task.Run(() => {
    foreach (var msg in taskQueue.GetConsumingEnumerable()) {
        // 处理消息逻辑
        Console.WriteLine("处理消息: {0}", msg);
    }
});

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