Aedes MQTT服务器事件同步响应与客户端消息顺序问题咨询

首先明确核心问题：Aedes会自动返回MQTT协议规定的ACK（如CONNACK、SUBACK），不会等待你在client/subscribe/publish事件中的自定义异步逻辑完成。这就是为什么getClient还没执行完，后续的subscribe/publish事件就被触发——因为客户端已经收到CONNACK，认为连接完成，立刻发送了后续请求。

以下是三种可行的解决方式：

方案1：服务器延迟返回ACK，等待自定义逻辑完成 #

这是最直接的根治方案，让客户端必须等服务器处理完前置逻辑后，再收到ACK并发送下一条请求。

针对client事件（客户端连接请求），手动拦截Aedes的默认CONNACK发送，等getClient完成后再主动返回：

aedes.on('client', async (client) => {
  // 阻止Aedes自动发送CONNACK
  client.connackSent = true;

  try {
    // 执行耗时的getClient逻辑
    await getClient(client.id);
    // 逻辑完成后，返回成功的CONNACK（返回码0表示连接允许）
    client.connack({ returnCode: 0 });
  } catch (err) {
    // 逻辑失败时，返回拒绝连接的CONNACK（返回码5）
    client.connack({ returnCode: 5 });
  }
});

同理，若需要拦截subscribe/publish的ACK，可采用类似逻辑：

• 订阅事件：用client.suback(subscriptions, grantedQoS)手动返回SUBACK
• QoS1/2的发布事件：用client.puback(packet)等方法手动返回对应ACK

方案2：客户端侧控制发送顺序 #

如果不想修改服务器逻辑，可在C++ Mosquitto客户端中，等待服务器发送的"就绪信号"后再发送后续请求：

服务器侧（Aedes） #

在getClient完成后，主动给客户端发送一个特定主题的就绪通知：

aedes.on('client', async (client) => {
  await getClient(client.id);
  // 向客户端发送就绪消息
  aedes.publish({
    topic: `/client/ready/${client.id}`,
    payload: 'ready',
    qos: 0,
    retain: false
  });
});

客户端侧（C++ Mosquitto） #

收到就绪通知后再执行订阅/发布操作：

#include <mosquitto.h>
#include <string>
#include <atomic>

std::atomic<bool> client_ready = false;
const char* CLIENT_ID = "my_client";
const char* READY_TOPIC = "/client/ready/my_client";

void on_message(struct mosquitto* mosq, void* obj, const struct mosquitto_message* msg) {
    std::string topic((char*)msg->topic);
    if (topic == READY_TOPIC) {
        client_ready = true;
        // 收到就绪信号后，发送订阅请求
        mosquitto_subscribe(mosq, NULL, "target/topic", 0);
    }
}

void on_connect(struct mosquitto* mosq, void* obj, int rc) {
    if (rc == 0) {
        // 订阅就绪通知主题
        mosquitto_subscribe(mosq, NULL, READY_TOPIC, 0);
    }
}

int main() {
    mosquitto_lib_init();
    struct mosquitto* mosq = mosquitto_new(CLIENT_ID, true, NULL);
    
    mosquitto_connect_callback_set(mosq, on_connect);
    mosquitto_message_callback_set(mosq, on_message);
    mosquitto_connect(mosq, "localhost", 1883, 60);
    
    while (mosquitto_loop(mosq, -1, 1) == MOSQ_ERR_SUCCESS) {
        // 可在此处理其他逻辑，或等待就绪后发送自定义消息
        if (client_ready) {
            // 示例：发送一条测试消息
            mosquitto_publish(mosq, NULL, "test/topic", 5, "hello", 0, false);
            client_ready = false; // 避免重复发送
        }
    }
    
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
    return 0;
}

方案3：服务器侧维护客户端状态，过滤提前触发的事件 #

在服务器侧用一个状态表记录客户端是否完成前置逻辑，未就绪时直接拒绝后续请求：

// 存储客户端就绪状态
const clientReadyStates = new Map();

aedes.on('client', async (client) => {
    clientReadyStates.set(client.id, false);
    try {
        await getClient(client.id);
        clientReadyStates.set(client.id, true);
    } catch (err) {
        clientReadyStates.set(client.id, false);
        client.disconnect(); // 逻辑失败直接断开连接
    }
});

aedes.on('subscribe', (client, subscriptions, callback) => {
    const isReady = clientReadyStates.get(client.id);
    if (!isReady) {
        callback(new Error('Client not initialized'));
        return;
    }
    // 执行自定义订阅逻辑
    // ...
    // 返回SUBACK
    callback(null, subscriptions.map(sub => sub.qos));
});

aedes.on('publish', (client, packet, callback) => {
    // 匿名客户端（如QoS0消息）需额外判断
    if (!client) {
        callback(null);
        return;
    }
    const isReady = clientReadyStates.get(client.id);
    if (!isReady) {
        callback(new Error('Client not initialized'));
        return;
    }
    // 执行自定义发布逻辑
    // ...
    callback(null);
});

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