如何识别Rebus工作线程？基于Rebus 6.6.4.0与Rebus.RabbitMq 7.3.5.0

问题（翻译后） #

我正在使用 Rebus 6.6.4.0 和 Rebus.RabbitMq 7.3.5.0，尝试通过线程名称识别消息处理程序中的每个线程。原本以为线程默认命名为“Rebus 1 worker 1”这类，但实际发现很多线程无名称——因为它们来自线程池，可能带有任意 ManagedThreadId 且无预设名称。我的需求是识别某个工作线程，允许它处理特定消息类型，其他线程则执行 Failfast，请问是否有办法识别 Rebus 工作线程/线程？

要实现识别 Rebus 工作线程并限制特定线程处理指定消息类型的需求，可以通过自定义工作线程工厂结合**异步本地存储（AsyncLocal）**来实现，具体步骤如下：

1. 自定义工作线程工厂，标记工作线程 #

Rebus 允许通过 SetWorkerThreadFactory 替换默认的工作线程创建逻辑，我们可以在这里给每个工作线程设置名称，并通过 AsyncLocal 存储线程唯一标识（确保异步处理时标识不会丢失）。

示例代码：

// 全局静态存储工作线程编号
private static readonly AsyncLocal<int?> _currentWorkerId = new AsyncLocal<int?>();

// Rebus 配置
Configure.With(yourActivator)
    .Transport(t => t.UseRabbitMq("your-rabbitmq-connection-string", "your-queue-name"))
    .Options(o =>
    {
        // 设置工作线程数量，比如5个
        o.SetNumberOfWorkers(5);
        
        // 自定义工作线程工厂
        o.SetWorkerThreadFactory((workerIndex, workLoop) =>
        {
            var threadName = $"Rebus-Worker-{workerIndex + 1}";
            var workerId = workerIndex + 1;
            
            return new Thread(() =>
            {
                // 标记当前工作线程的ID
                _currentWorkerId.Value = workerId;
                Thread.CurrentThread.Name = threadName;
                
                try
                {
                    // 执行Rebus的工作循环
                    workLoop();
                }
                finally
                {
                    // 清理标识
                    _currentWorkerId.Value = null;
                }
            })
            {
                Name = threadName,
                IsBackground = true
            };
        });
    })
    .Start();

2. 在消息处理程序中验证线程标识 #

在目标消息的处理程序中，通过 AsyncLocal 获取当前工作线程ID，判断是否为允许处理的线程，否则执行 FailFast。

示例处理程序：

public class SpecificMessageHandler : IHandleMessages<SpecificMessage>
{
    public async Task Handle(SpecificMessage message, IMessageContext context)
    {
        // 获取当前工作线程ID
        var currentWorkerId = _currentWorkerId.Value;
        
        // 只允许ID为1的工作线程处理该消息
        if (currentWorkerId != 1)
        {
            Environment.FailFast($"Worker thread {currentWorkerId} is not authorized to handle SpecificMessage.");
        }

        // 正常处理消息逻辑
        await ProcessSpecificMessage(message);
    }

    private async Task ProcessSpecificMessage(SpecificMessage message)
    {
        // 你的业务逻辑
    }
}

3. 可选：通过管道拦截器全局检查 #

如果需要对多个消息类型统一做线程校验，可以实现 Rebus 的 IIncomingStep 管道拦截器，在消息进入处理程序前统一验证：

public class WorkerThreadAuthorizationStep : IIncomingStep
{
    public async Task Process(IncomingStepContext context, Func<Task> next)
    {
        var message = context.Load<Message>();
        
        // 针对指定消息类型做校验
        if (message.Message is SpecificMessage || message.Message is AnotherRestrictedMessage)
        {
            var currentWorkerId = _currentWorkerId.Value;
            if (currentWorkerId != 1)
            {
                Environment.FailFast($"Worker thread {currentWorkerId} cannot handle restricted message type.");
            }
        }

        // 继续执行后续处理步骤
        await next();
    }
}

然后在 Rebus 配置中注册该拦截器：

Options(o =>
{
    o.AddIncomingStep<WorkerThreadAuthorizationStep>();
    // 其他配置...
})

注意事项 #

• Environment.FailFast 会直接终止进程，使用前需确认业务场景是否允许这种极端行为——如果只是想拒绝处理而非终止进程，可以改为抛出异常让 Rebus 重试或转入错误队列。
• Rebus 默认工作线程会被命名，但如果处理程序中存在异步操作，后续代码可能切换到线程池的无名线程，因此 AsyncLocal 比线程名称更可靠。
• 自定义线程工厂时，确保设置 IsBackground = true，避免工作线程阻止进程退出。

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