使用MqttNet接收MQTT消息时如何正确触发异步处理流程？

我们在远程服务器部署了一组.NET Core控制台应用，通过MqttNet接收MQTT消息。客户端收到消息后，以消息主题为键匹配预先注册的Action分发执行，初始客户端启动代码如下：

public void StartClient()
{
    mqttClient.ApplicationMessageReceived += (s, e) =>
    {
        var clientId = e.ClientId;
        var topic = e.ApplicationMessage.Topic;
        var payload = e.ApplicationMessage.ConvertPayloadToString();

        Console.WriteLine("");
        Console.WriteLine($"ClientId: {clientId}");
        Console.WriteLine($"Topic: {topic}");
        Console.WriteLine($"Payload: {payload}");

        var action = actions[topic];

        action?.Invoke(payload);
    };

    mqttClient.StartAsync(options).Wait();
}

同步Action的典型场景为修改内存状态标志位，比如收到启动完成消息时的处理：

public void StartupCompleted(string message)
{
    IsInStartup = false;
}

这类短耗时同步Action执行无异常，但触发异步或长耗时流程时，.NET Core应用会出现崩溃。当前核心业务需求是将MQTT收到的消息路由转发至Azure Event Hubs：

• 消息量较小时，通过Task.Run()实现即发即弃逐条发布，运行正常
• 消息量增长后，Azure端因负载过高开始拒绝消息，由于发布逻辑包裹在Task.Run()中，异常被完全丢失，无法感知发布失败，也无法执行重试调度

待确认的问题：

1. 是否可以在顶层ApplicationMessageReceived方法中注册各消息主题对应的Task委托，收到消息时调用执行？
2. 该实现是否会阻塞ApplicationMessage消息处理流程，直到关联Task执行完毕？

之前尝试的两种方案均存在问题：

• 方案1：将Azure异步方法强制转为同步执行，通过AsyncContext捕获异常，再以Fire and Forget方式触发Action
  EventHub包装类代码：

  public void PublishEvents(string connectionString, IEnumerable<AzureEventMessage> azureEventMessages)
  {
      var task = Task.Run(() => AsyncContext.Run(() => PublishEventsAsync(connectionString, azureEventMessages)));
      task.GetAwaiter().GetResult();
  }
  

  注册的Action代码：

  public void PublishToAzure(string message)
  {
      Task.Run(() => eventService.PublishToAzure(message));
  }
  

• 方案2：保留异步链路，Action中直接通过Task.Run运行异步方法

  public void PublishToAzure(string message)
  {
      Task.Run(async () => await eventService.PublishToAzure(message));
  }
  

  该方案额外存在消息丢失问题：如果ApplicationMessageReceived处理程序需要在下一条消息到达前执行完成，即使将内部队列长度从默认250条调整到5000-10000条，随时间推移队列依然会被填满。

关于异步委托注册的可行性 #

可以注册Func<string, Task>类型的异步消息处理委托，但需要注意事件处理器的正确写法，直接在原同步匿名方法中写await会触发编译错误，需要将匿名方法标记为async，修正后的代码如下：

// 注意：该写法会阻塞消息接收流程，非特殊场景不推荐
mqttClient.ApplicationMessageReceived += async (s, e) =>
{
    var clientId = e.ClientId;
    var topic = e.ApplicationMessage.Topic;
    var payload = e.ApplicationMessage.ConvertPayloadToString();

    Console.WriteLine("");
    Console.WriteLine($"ClientId: {clientId}");
    Console.WriteLine($"Topic: {topic}");
    Console.WriteLine($"Payload: {payload}");

    if (tasks.TryGetValue(topic, out var handler))
    {
        // async void 方法内必须加全局异常捕获，否则未处理异常会直接导致进程崩溃
        try
        {
            await handler(payload);
        }
        catch (Exception ex)
        {
            Console.WriteLine($"处理主题{topic}消息出错：{ex}");
        }
    }
};

mqttClient.StartAsync(options).Wait();

关于是否阻塞消息处理流程 #

该写法确实会阻塞MQTT消息处理流程。MqttNet默认采用单线程串行模式派发消息，上一个消息的回调执行完成前，不会触发下一个消息的接收处理。如果在回调中await长耗时的EventHub发布逻辑，消息消费速度会被严重拖慢，不管把内部接收队列调多大，消息量上来后迟早会被填满，最终出现丢消息问题。

原有方案的问题说明 #

• 强制同步阻塞异步方法的写法：混合Task.Run、AsyncContext.Run再同步等待属于典型的异步反模式，不仅会增加不必要的线程切换开销，还存在线程池耗尽风险，外层套Task.Run做即发即弃依然会丢失异常，没有解决核心问题。
• 直接Task.Run跑异步逻辑的写法：属于标准的即发即弃模式，只要不await返回的Task，也不在Task内部做异常捕获，Task中抛出的异常（比如EventHub限流、连接失败）会被线程池吞掉，既无感知也无法重试，未捕获的异常还可能在GC回收Task时触发进程崩溃。

采用接收与处理解耦的架构，从根源上避免回调阻塞、异常丢失、队列溢出问题：

1. 拆分消息处理逻辑：短耗时同步操作（比如修改内存状态位）保留原同步Action逻辑；长耗时异步操作（比如转发EventHub）统一走后台队列处理。
2. 使用高性能队列做缓冲：用.NET Core自带的System.Threading.Channels.Channel作为本地消息缓冲队列，MQTT接收回调只做日志打印、消息入队操作，不执行任何长耗时逻辑，入队完成立刻返回，完全不阻塞MqttNet的消息接收。
3. 独立后台服务消费队列：启动单独的后台任务消费队列中的消息，匹配对应异步处理委托执行，统一在消费逻辑中做异常捕获、重试、限流、批量处理。

简化实现代码如下：

// 初始化有界消息队列，设置最大容量，队列满时采用等待策略实现背压，避免丢消息
private readonly Channel<(string Topic, string Payload)> _messageQueue = Channel.CreateBounded<(string, string)>(new BoundedChannelOptions(10000)
{
    FullMode = BoundedChannelFullMode.Wait
});

public void StartClient()
{
    // 消息接收回调仅做最轻量的入队操作
    mqttClient.ApplicationMessageReceived += (s, e) =>
    {
        var clientId = e.ClientId;
        var topic = e.ApplicationMessage.Topic;
        var payload = e.ApplicationMessage.ConvertPayloadToString();

        Console.WriteLine("");
        Console.WriteLine($"ClientId: {clientId}");
        Console.WriteLine($"Topic: {topic}");
        Console.WriteLine($"Payload: {payload}");

        // 队列满时等待，不会丢消息
        _ = _messageQueue.Writer.WriteAsync((topic, payload));
    };

    mqttClient.StartAsync(options).Wait();

    // 启动后台消息消费任务
    _ = ConsumeMessagesAsync();
}

private async Task ConsumeMessagesAsync()
{
    // 持续从队列读取消息处理
    await foreach (var msg in _messageQueue.Reader.ReadAllAsync())
    {
        // 先匹配同步Action，存在则直接执行
        if (actions.TryGetValue(msg.Topic, out var syncHandler))
        {
            try
            {
                syncHandler(msg.Payload);
            }
            catch (Exception ex)
            {
                Console.WriteLine($"同步处理主题{msg.Topic}消息出错：{ex}");
            }
            continue;
        }

        // 匹配异步处理委托
        if (tasks.TryGetValue(msg.Topic, out var asyncHandler))
        {
            try
            {
                await asyncHandler(msg.Payload);
            }
            catch (Exception ex)
            {
                // 统一处理异常：记录日志、指数退避重试、投递死信队列均可在此实现
                Console.WriteLine($"异步处理主题{msg.Topic}消息出错：{ex}");
            }
        }
    }
}

该方案的优势：

• MQTT接收回调无长耗时逻辑，不会阻塞MqttNet内部消息派发，从根源上避免内部接收队列被填满
• 所有处理逻辑的异常均可统一捕获，不会出现异常丢失、无感知失败的问题
• 可灵活调整消费并发度、添加批量发布逻辑，适配EventHub的限流场景，大幅提升消息吞吐量
• 队列满时的等待策略可实现天然背压，避免内存溢出问题

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