如何结合Rebus的Bus.Defer功能与指定队列发送消息？

我之前刚好在项目里处理过几乎一模一样的场景，给你分享下我的实现思路和具体代码，应该能解决你的问题~

核心思路 #

Rebus的默认Defer方法会把延迟消息到期后发送到当前端点的输入队列，但你的需求是要转发到指定的queue1/queue2，所以关键是要把目标队列信息和原始消息一起传递，等延迟到期后，再通过高级路由把消息发送到目标队列。

步骤1：创建延迟消息包装类（推荐方式） #

为了清晰区分普通消息和延迟消息，同时携带目标队列信息，我会创建一个通用的包装类：

public class DeferredMessageWrapper<T>
{
    // 原始业务消息
    public T BusinessMessage { get; set; }
    // 到期后要发送到的目标队列名
    public string TargetQueueName { get; set; }
}

步骤2：延迟发送包装后的消息 #

当你需要延迟发送到指定队列时，把原始消息包装起来，调用Defer方法（如果还没实现外部超时管理器，暂时可以用Rebus内置的超时机制，后续替换即可）：

// 假设你的业务消息是YourBusinessMessage类型
var businessMsg = new YourBusinessMessage { /* 消息内容 */ };

// 包装消息，指定目标队列
var deferredMsg = new DeferredMessageWrapper<YourBusinessMessage>
{
    BusinessMessage = businessMsg,
    TargetQueueName = "queue1" // 按需替换成"queue2"
};

// 延迟5分钟发送（时间按需调整）
await bus.Defer(TimeSpan.FromMinutes(5), deferredMsg);

步骤3：实现包装消息的处理Handler #

这个Handler的作用就是收到延迟到期的包装消息后，通过高级路由把原始业务消息发送到指定队列：

public class DeferredMessageWrapperHandler<T> : IHandleMessages<DeferredMessageWrapper<T>>
{
    private readonly IBus _bus;

    public DeferredMessageWrapperHandler(IBus bus)
    {
        _bus = bus;
    }

    public async Task Handle(DeferredMessageWrapper<T> message)
    {
        // 用高级路由发送到目标队列
        await _bus.Advanced.Routing.Send(message.TargetQueueName, message.BusinessMessage);
    }
}

步骤4：后续切换到外部超时管理器 #

等你实现外部超时管理器后，只需要在Rebus配置中指定使用外部超时队列即可，代码逻辑不需要大改：

Configure.With(new CastleWindsorContainerAdapter(container))
    .Transport(t => t.UseRabbitMq("your-rabbitmq-connection-string", "your-handler-endpoint-name"))
    .Timeouts(t => t.UseExternalTimeoutManager("rebus-external-timeout-manager-queue")) // 指定外部超时队列
    .Routing(r => r.TypeBased().MapAssemblyOf<YourBusinessMessage>("your-main-endpoint"))
    .Start();

此时延迟的包装消息会先发送到外部超时管理器队列，到期后再转发到你的Handler所在端点，Handler再把业务消息路由到queue1/queue2。

备选方案：通过消息头传递目标队列（不推荐） #

如果你不想用包装类，也可以通过自定义消息头传递目标队列信息：

var headers = new Dictionary<string, string>
{
    { "X-Target-Queue", "queue1" }
};

// 延迟发送时携带头信息
await bus.Defer(TimeSpan.FromMinutes(5), businessMsg, headers);

然后在处理延迟消息的Handler中（注意要确保这个Handler只处理延迟后的消息，避免和普通业务消息混淆）：

public class DeferredBusinessMessageHandler : IHandleMessages<YourBusinessMessage>
{
    private readonly IBus _bus;
    private readonly IMessageContext _messageContext;

    public DeferredBusinessMessageHandler(IBus bus, IMessageContext messageContext)
    {
        _bus = bus;
        _messageContext = messageContext;
    }

    public async Task Handle(YourBusinessMessage message)
    {
        if (_messageContext.Headers.TryGetValue("X-Target-Queue", out var targetQueue))
        {
            await _bus.Advanced.Routing.Send(targetQueue, message);
        }
        // 这里要注意：如果是普通业务消息，不要处理，避免重复消费
        else
        {
            // 可以抛出异常或者忽略，根据你的业务逻辑调整
            throw new InvalidOperationException("Received non-deferred business message in deferred handler");
        }
    }
}

不过这种方式容易和普通业务消息混淆，所以还是推荐用包装类的方式，更清晰可靠。

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