如何在Rebus消息处理程序中立即停止处理新消息？

核心问题原因 #

• 此前在AfterMessageHandled阶段执行停止逻辑时机过晚：Rebus会预拉取少量消息到内存缓冲区排队，触发停止时已经有额外消息进入待处理队列，会继续被执行
• 在处理程序内同步调用bus.Advanced.Workers.SetNumberOfWorkers(0)必然死锁：该方法会阻塞等待所有活跃处理程序执行完成，调用方本身就是正在运行的处理程序，形成循环等待

正确实现方案 #

你需要组合使用流水线最前置的拦截钩子+异步触发停止逻辑两个手段解决问题：

步骤1：定义全局线程安全停止标志 #

先声明一个全局的volatile布尔变量作为停止开关，避免多线程读写的可见性问题：

private volatile bool _stopBusProcessing = false;

步骤2：注册BeforeReceiveMessage事件拦截 #

BeforeReceiveMessage是Rebus消息处理流水线的最前端钩子，会在每次尝试从队列服务拉取新消息之前触发，在这个节点做拦截可以从根源上阻止新消息进入处理流程：

bus.Advanced.Events.BeforeReceiveMessage += (sender, args) =>
{
    if (_stopBusProcessing)
    {
        // 取消本次消息拉取操作
        args.Cancel = true;
    }
    return Task.CompletedTask;
};

如果需要完全杜绝已经预拉取到内存的消息被处理，可以额外注册BeforeMessageHandled事件做二层拦截：

bus.Advanced.Events.BeforeMessageHandled += (sender, args) =>
{
    if (_stopBusProcessing)
    {
        // 跳过当前消息的处理逻辑
        args.SkipInvocation = true;
    }
    return Task.CompletedTask;
};

步骤3：异步触发总线停止逻辑 #

当满足异常停止或业务停止条件时，先修改停止标志，再将停止工作线程的逻辑放到独立的线程池任务中执行，不要阻塞当前处理线程，规避死锁问题：

// 你的消息处理逻辑
try
{
    // 业务处理代码
    if(/* 满足业务停止条件 */)
    {
        StopBus();
    }
}
catch (Exception ex)
{
    // 异常处理逻辑
    StopBus();
}

// 停止总线的封装方法
void StopBus()
{
    // 先设停止标志，立刻拦截所有新消息拉取
    _stopBusProcessing = true;
    // 异步执行工作线程停止逻辑，不等待返回避免死锁
    _ = Task.Run(() => bus.Advanced.Workers.SetNumberOfWorkers(0));
}

效果说明 #

单工作线程、并行度为1的配置下，这套方案触发停止后最多只会多处理1条已经处于执行中的消息，不会有更多后续消息被拉取或处理，同时完全规避了死锁问题。

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