使用IReceiveObserver时ConsumeContext<T>发布消息不生效问题

数据老陈

2026-9-2

问题根因

故障出现是三个核心逻辑缺失/理解偏差导致的：

未等待异步Publish执行完成：ConsumeContext.Publish()是异步方法，返回Task对象，你现有代码直接调用不做await等待，ConsumeFault方法会立刻返回，此时消费管道正处于故障回收阶段，会直接释放上下文资源，Publish操作还没实际执行就被中断，消息自然发不出去。
发布操作被绑定在原消费的失败事务上下文中：IReceiveObserver触发故障回调时，当前传入的ConsumeContext和原消费流程共享同一个作用域与事务边界。一旦原消费判定为失败（要进入重试、死信队列），这个作用域下所有未提交的发送、发布操作都会随事务回滚被撤销，不会真正投递到总线。
部分总线级故障场景下传入的上下文本身不完整：比如序列化错误、消息格式不合法这类故障发生时，框架还没完成消息体的反序列化，此时上下文中的Message对象是空值、发送端点配置未初始化，直接用这个上下文发布根本没有可执行的发送链路。

修复方案

按下面的逻辑调整即可解决：

所有异步操作必须显式等待，禁止丢弃Task对象不处理。
不要复用故障回调传入的ConsumeContext做发布操作，而是通过构造函数注入独立的IBus实例，用总线实例发布故障消息，脱离原消费的事务边界，发布操作不会随原消费失败回滚。
参考实现代码：

public class JobStatusReceiveObserver : IReceiveObserver
{
    private readonly IBus _bus;
    // 构造注入总线实例，和消费上下文生命周期隔离
    public JobStatusReceiveObserver(IBus bus)
    {
        _bus = bus;
    }

    public async Task ConsumeFault<T>(ConsumeContext<T> context, TimeSpan elapsed, string consumerType, Exception exception) where T : class
    {
        // 提取业务标识，注意序列化失败场景下T不是强类型业务消息，要从消息头取JobId
        var jobId = context.Headers.TryGetHeader("JobId", out var headerVal) 
            ? Guid.Parse(headerVal.ToString()!) 
            : (context.Message is JobMessage msg ? msg.JobId : Guid.Empty);
        
        // 用独立总线实例发布，必须await
        await _bus.Publish(new JobStatusUpdatedEvent
        {
            JobId = jobId,
            Status = JobStatus.Failed,
            ErrorMessage = exception.Message,
            ConsumeElapsedMs = elapsed.TotalMilliseconds
        });
    }

    // 其余IReceiveObserver接口方法（PreReceive、PostReceive、ReceiveFault、ReceiveEndpointFault等）按需求实现即可
}

补充幂等判断：如果配置了消费重试，每次重试失败都会触发ConsumeFault回调，要加判断（比如判断是否为最后一次重试失败、或者事件携带唯一标识做去重），避免重复推送故障状态导致看板数据异常。

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