如何在EasyNetQ的PubSub中仅配置同步阻塞调用？

要实现100%同步阻塞的消息处理（逐个顺序处理，无并发），你需要从RabbitMQ预取配置、EasyNetQ订阅方式、避免重复订阅三个核心点调整，以下是具体方案：

核心问题分析 #

你遇到的0.8%异步/并发场景，大概率源于两个问题：

• 未限制RabbitMQ的消息预取数量，导致消费者提前获取多条消息；
• 重连事件中重复订阅同一subscriptionId，创建了多个消费者实例，引发并发处理。

具体配置步骤 #

1. 强制设置预取计数为1 #

通过配置WithPrefetchCount(1)，让RabbitMQ每次只给消费者推送1条消息，必须等当前消息处理完成并ACK后，才会推送下一条。这是实现同步阻塞的基础。

2. 使用同步Subscribe方法并显式限制消费者数量 #

放弃SubscribeAsync，改用同步Subscribe方法，并通过WithConsumerCount(1)确保同一订阅ID下只有1个消费者实例，彻底杜绝并发。

3. 修复重连时的重复订阅问题 #

你的Advanced_Connected事件中再次调用SubscribeAsync，会导致同一subscriptionId下重复创建消费者，这是并发的关键诱因。改为在重连时先取消原有订阅，再重新订阅，避免重复实例。

修改后的代码示例 #

public class Subscriber : ISubscriber, IDisposable
{
    private readonly IBus _bus;
    private readonly IMessageProcessor _myMessageProcessor;
    private readonly ILogger<ISubscriber> _logger;
    private const string _subscriptionId = "MySubs";
    private IDisposable _subscription; // 保存订阅句柄，用于重连时取消

    public Subscriber(IBus bus, IMessageProcessor myMessageProcessor, ILogger<ISubscriber> logger)
    {
        _bus = bus;
        _myMessageProcessor = myMessageProcessor;
        _logger = logger;
        _bus.Advanced.Connected += Advanced_Connected;
    }

    public void Subscribe()
    {
        CreateSubscription();
    }

    private void CreateSubscription()
    {
        // 先取消原有订阅（如果存在）
        _subscription?.Dispose();
        
        // 同步订阅，配置预取计数1、消费者数量1
        _subscription = _bus.PubSub.Subscribe<MyCustomMessage>(
            _subscriptionId,
            OnMyCustomMessage,
            c => c.WithAutoDelete(false)
                  .WithPrefetchCount(1)
                  .WithConsumerCount(1)
        );
    }

    private void Advanced_Connected(object sender, EventArgs e)
    {
        // 重连时重新创建订阅（先取消旧的）
        CreateSubscription();
    }

    private void OnMyCustomMessage(MyCustomMessage message)
    {
        try {
            if (_logger.IsEnabled(LogLevel.Debug))
                _logger.LogDebug($"Received message #{message.Id}. {Newtonsoft.Json.JsonConvert.SerializeObject(message)}");
            _myMessageProcessor.ProcessCustomMessage(message);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, $"{nameof(Subscriber)}.{nameof(OnMyCustomMessage)}: args: {Newtonsoft.Json.JsonConvert.SerializeObject(message)}");
        }
    }

    public void Dispose()
    {
        _subscription?.Dispose();
        _bus.Advanced.Connected -= Advanced_Connected;
    }
}

额外验证点 #

• 确认IMessageProcessor.ProcessCustomMessage是完全同步阻塞的方法，内部没有异步调用（比如await、Task.Run等）；
• 检查EasyNetQ全局配置，确保没有被覆盖的消费者并发设置；
• 登录RabbitMQ管理后台，确认该队列的消费者数量始终为1，预取计数为1。

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