MQTT消息送达ACK实现问题：事件监听无法终止

咱们来拆解下你遇到的问题：核心是两个点——绑定的message事件监听器没法自动停止，还有你在发完消息后立刻就断开了连接，大概率还没等到ACK就结束流程了。咱们一步步来修复：

问题根源分析 #

• 你在publish的回调里用client.on('message', ...)绑定了持久化的事件监听器，这个监听器会一直挂着，除非你手动移除，哪怕收到目标ACK后，它还会继续监听后续所有消息。
• 代码里publish之后立刻await client.end()，直接关闭了MQTT连接，这会导致还没等接收方返回ACK，整个流程就终止了。

针对性解决方案 #

1. 用一次性监听器或者手动移除监听器 #

最省心的方式是用client.once('message', ...)，它会在触发一次后自动移除自己，完美匹配ACK只需要接收一次的场景。如果业务需要更灵活的控制，也可以把监听器函数存起来，匹配到目标ACK后用client.off('message', 监听器函数)手动移除。

2. 等ACK接收完成再关闭连接 #

绝对不能在发完消息后立刻断开连接，得等确认收到对应ID的ACK后，再执行关闭操作。

修改后的完整代码 #

let mqtt = require('async-mqtt'), cfg = require('./cfg');
let client = mqtt.connect(cfg.server);
let id = 123;

async function sendWithAck() {
  try {
    // 先完成ACK主题的订阅（确保发消息前订阅生效，避免丢ACK）
    await client.subscribe('some/other/topic');
    
    // 发送带ID的消息
    await client.publish('some/topic', `Message with id${id}`);
    
    // 用Promise封装一次性事件监听，等待目标ACK
    await new Promise((resolve) => {
      client.once('message', function listener(topic, msg) {
        const receivedContent = msg.toString();
        console.log(`${topic}> ${receivedContent}`);
        
        // 校验是否是当前消息对应的ACK
        if (receivedContent.includes(`id${id}`)) {
          console.log(`成功收到ID为${id}的消息ACK`);
          resolve();
        } else {
          // 如果收到的是其他消息，重新绑定一次性监听（根据你的业务需求调整）
          client.once('message', listener);
        }
      });
    });
    
    // 确认收到ACK后再关闭连接
    await client.end();
    console.log('流程完成');
  } catch(e) {
    console.log('出错了', e);
    process.exit();
  }
}

client.on('connect', sendWithAck);

额外优化建议 #

如果你的ACK是JSON格式（更规范的做法），可以解析成对象后精准匹配ID，比如：

const ackInfo = JSON.parse(msg.toString());
if (ackInfo.msgId === id) {
  // 处理ACK逻辑
}

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