异步MQTT 客户端库for C

基于C语言编写的MQTTAsync API，版权属于IBM，适用于2009年至2018年。

Paho MQTT C库中的API分为两类：

• 同步API（MQTTClient API）
  • 特点：被认为是更易于使用，部分函数调用会导致程序暂停执行（即阻塞），直到操作完成。这意味着在等待操作结果时，程序不会继续执行其他任务。
  • 线程安全：此API不是线程安全的，意味着在多线程环境下直接使用可能会遇到竞态条件等问题，需要开发者自己处理同步问题。
• 异步API（MQTTAsync API）：
  • 特点：设计为完全非阻塞的，所有调用都不会导致程序暂停，非常适合那些需要高度响应性的应用场景，特别是在有图形用户界面(GUI)的环境中，因为它能确保UI始终保持流畅。
  • 线程安全：与同步API不同，异步API是线程安全的，意味着它可以在多线程程序中安全地使用，而不需要额外的同步措施。

本文章介绍的是MQTTAsync API
MQTT客户端应用程序连接到支持MQTT的服务器。典型的客户端负责从遥测设备收集信息并将信息发布到服务器。它还可以订阅主题、接收消息，并使用这些信息控制遥测设备。
MQTT客户端实现已发布的MQTT v3协议。您可以使用自己选择的编程语言和平台为MQTT协议编写自己的API。这可能耗时且容易出错。
为了简化编写MQTT客户端应用程序，此库为您封装了MQTT v3协议。使用此库可以用几行代码编写功能齐全的MQTT客户端应用程序。这里提供的信息记录了异步MQTT客户端库为C提供的API。

使用MQTT客户端库开发应用程序

使用MQTT客户端库开发应用程序时，通常会遵循一个相似的结构步骤，以确保高效、有序地与MQTT服务器交互。下面是这一过程的详细说明：

1. 创建客户端对象： 首先，你需要实例化一个客户端对象。这一步骤实际上是为你的应用程序创建一个MQTT客户端实例，它是与MQTT服务器通信的基础。
2. 设置连接选项： 接下来，配置客户端以便连接到MQTT服务器。这包括设置服务器地址、端口、客户端ID、用户名、密码等连接参数。此外，还可以设置诸如超时时间、清洁会话标志等高级选项。
3. 设置回调函数： 为了处理来自服务器的各种事件（如消息到达、连接状态变化等），你需要定义并设置回调函数。这些函数会在相应的事件发生时被库自动调用，让你的程序能够及时响应。
4. 连接到MQTT服务器： 使用之前设置的选项调用连接函数，尝试建立与MQTT服务器的连接。这一步可能成功也可能失败，具体取决于网络状况、服务器状态以及提供的凭据是否有效。
5. 订阅主题： 一旦连接成功，客户端可以订阅它感兴趣的MQTT主题。订阅后，当有消息发布到这些主题时，客户端就会收到通知并通过之前设置的回调函数处理这些消息。
6. 消息循环： 进入主循环，此阶段客户端主要执行以下操作：
   • 发布消息：根据需要，客户端可以发布消息到特定的主题上，实现与其它客户端的数据共享。
   • 处理接收到的消息：通过回调函数处理从服务器接收到的所有消息。
     监听并处理其他事件，如连接断开、重连等。
7. 断开连接： 当应用程序完成其任务或需要终止时，应优雅地断开与MQTT服务器的连接。这通常涉及调用客户端库提供的断开函数。
8. 释放资源： 最后，确保释放由客户端对象占用的所有资源，包括内存和其他系统资源。这是良好的编程实践，有助于防止内存泄漏和其他资源管理问题。

遵循这一结构，可以确保你的MQTT客户端应用程序逻辑清晰、易于维护，并能有效地与MQTT服务器交互。

示例

下面是对提到的几个简单示例：
环境搭建和编译参考MQTT C Client for Posix and Windows

发布

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTAsync.h"
#define ADDRESS     "tcp://localhost:1883"
#define CLIENTID    "ExampleClientPub"
#define TOPIC       "hellotopic/5"
#define PAYLOAD     "Hello World!"
#define QOS         1
#define TIMEOUT     10000L
volatile MQTTAsync_token deliveredtoken;
int finished = 0;
void connlost(void *context, char *cause)
{
        MQTTAsync client = (MQTTAsync)context;
        MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
        int rc;
        printf("\nConnection lost\n");
        printf("     cause: %s\n", cause);
        printf("Reconnecting\n");
        conn_opts.keepAliveInterval = 20;
        conn_opts.cleansession = 1;
        if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
        {
                printf("Failed to start connect, return code %d\n", rc);
                finished = 1;
        }
}
void onDisconnect(void* context, MQTTAsync_successData* response)
{
        printf("Successful disconnection\n");
        finished = 1;
}
void onSend(void* context, MQTTAsync_successData* response)
{
        MQTTAsync client = (MQTTAsync)context;
        MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
        int rc;
        printf("Message with token value %d delivery confirmed\n", response->token);
        opts.onSuccess = onDisconnect;
        opts.context = client;
        if ((rc = MQTTAsync_disconnect(client, &opts)) != MQTTASYNC_SUCCESS)
        {
                printf("Failed to start sendMessage, return code %d\n", rc);
                exit(EXIT_FAILURE);
        }
}
void onConnectFailure(void* context, MQTTAsync_failureData* response)
{
        printf("Connect failed, rc %d\n", response ? response->code : 0);
        finished = 1;
}
void onConnect(void* context, MQTTAsync_successData* response)
{
        MQTTAsync client = (MQTTAsync)context;
        MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
        MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
        int rc;
        printf("Successful connection\n");
        opts.onSuccess = onSend;
        opts.context = client;
        pubmsg.payload = PAYLOAD;
        pubmsg.payloadlen = strlen(PAYLOAD);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        deliveredtoken = 0;
        if ((rc = MQTTAsync_sendMessage(client, TOPIC, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
        {
                printf("Failed to start sendMessage, return code %d\n", rc);
                exit(EXIT_FAILURE);
        }
}
int main(int argc, char* argv[])
{
        MQTTAsync client;
        MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
        MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
        MQTTAsync_token token;
        int rc;
        MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
        MQTTAsync_setCallbacks(client, NULL, connlost, NULL, NULL);
        conn_opts.keepAliveInterval = 20;
        conn_opts.cleansession = 1;
        conn_opts.onSuccess = onConnect;
        conn_opts.onFailure = onConnectFailure;
        conn_opts.context = client;
        if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
        {
                printf("Failed to start connect, return code %d\n", rc);
                exit(EXIT_FAILURE);
        }
        printf("Waiting for publication of %s\n"
         "on topic %s for client with ClientID: %s\n",
         PAYLOAD, TOPIC, CLIENTID);
        while (!finished)
                #if defined(WIN32) || defined(WIN64)
                        Sleep(100);
                #else
                        usleep(10000L);
                #endif
        MQTTAsync_destroy(&client);
        return rc;
}

订阅

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTAsync.h"
#define ADDRESS     "tcp://localhost:1883"
#define CLIENTID    "ExampleClientSub"
#define TOPIC       "hellotopic/#"
#define PAYLOAD     "Hello World!"
#define QOS         1
#define TIMEOUT     10000L
volatile MQTTAsync_token deliveredtoken;
int disc_finished = 0;
int subscribed = 0;
int finished = 0;
void connlost(void *context, char *cause)
{
        MQTTAsync client = (MQTTAsync)context;
        MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
        int rc;
        printf("\nConnection lost\n");
        printf("     cause: %s\n", cause);
        printf("Reconnecting\n");
        conn_opts.keepAliveInterval = 20;
        conn_opts.cleansession = 1;
        if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
        {
                printf("Failed to start connect, return code %d\n", rc);
            finished = 1;
        }
}
int msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
    int i;
    char* payloadptr;
    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");
    payloadptr = message->payload;
    for(i=0; i<message->payloadlen; i++)
    {
        putchar(*payloadptr++);
    }
    putchar('\n');
    MQTTAsync_freeMessage(&message);
    MQTTAsync_free(topicName);
    return 1;
}
void onDisconnect(void* context, MQTTAsync_successData* response)
{
        printf("Successful disconnection\n");
        disc_finished = 1;
}
void onSubscribe(void* context, MQTTAsync_successData* response)
{
        printf("Subscribe succeeded\n");
        subscribed = 1;
}
void onSubscribeFailure(void* context, MQTTAsync_failureData* response)
{
        printf("Subscribe failed, rc %d\n", response ? response->code : 0);
        finished = 1;
}
void onConnectFailure(void* context, MQTTAsync_failureData* response)
{
        printf("Connect failed, rc %d\n", response ? response->code : 0);
        finished = 1;
}
void onConnect(void* context, MQTTAsync_successData* response)
{
        MQTTAsync client = (MQTTAsync)context;
        MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
        MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
        int rc;
        printf("Successful connection\n");
        printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n", TOPIC, CLIENTID, QOS);
        opts.onSuccess = onSubscribe;
        opts.onFailure = onSubscribeFailure;
        opts.context = client;
        deliveredtoken = 0;
        if ((rc = MQTTAsync_subscribe(client, TOPIC, QOS, &opts)) != MQTTASYNC_SUCCESS)
        {
                printf("Failed to start subscribe, return code %d\n", rc);
                exit(EXIT_FAILURE);
        }
}
int main(int argc, char* argv[])
{
        MQTTAsync client;
        MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
        MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
        MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
        MQTTAsync_token token;
        int rc;
        int ch;
        MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
        MQTTAsync_setCallbacks(client, NULL, connlost, msgarrvd, NULL);
        conn_opts.keepAliveInterval = 20;
        conn_opts.cleansession = 1;
        conn_opts.onSuccess = onConnect;
        conn_opts.onFailure = onConnectFailure;
        conn_opts.context = client;
        if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
        {
                printf("Failed to start connect, return code %d\n", rc);
                exit(EXIT_FAILURE);
        }
        while   (!subscribed)
                #if defined(WIN32) || defined(WIN64)
                        Sleep(100);
                #else
                        usleep(10000L);
                #endif
        if (finished)
                goto exit;
        do
        {
                ch = getchar();
        } while (ch!='Q' && ch != 'q');
        disc_opts.onSuccess = onDisconnect;
        if ((rc = MQTTAsync_disconnect(client, &disc_opts)) != MQTTASYNC_SUCCESS)
        {
                printf("Failed to start disconnect, return code %d\n", rc);
                exit(EXIT_FAILURE);
        }
        while   (!disc_finished)
                #if defined(WIN32) || defined(WIN64)
                        Sleep(100);
                #else
                        usleep(10000L);
                #endif
exit:
        MQTTAsync_destroy(&client);
        return rc;
}

重要概念补充说明

线程Threading

在使用MQTT异步客户端库开发多线程应用程序时，理解库的线程安全性和回调机制至关重要。该库设计为线程安全，意味着多个应用线程可以同时调用API函数而不会引发冲突。以下是关于如何在多线程环境中使用此库的一些建议和注意事项：

1. 后台处理与网络连接：握手处理和网络连接的维护工作在后台线程中执行，这使得应用的主线程可以继续执行其他任务而不被阻塞。
2. 线程安全的API：库的API函数设计为可以被多个应用线程安全调用。这意味着你可以放心地在不同的线程中调用如MQTTAsync_publish()、MQTTAsync_subscribe()等函数。
3. 回调机制：库使用回调函数来通知应用程序状态变更（如连接丢失、消息到达、消息发送完成等）。你需要通过MQTTAsync_setCallbacks()函数注册这些回调函数（例如MQTTAsync_messageArrived, MQTTAsync_connectionLost, MQTTAsync_deliveryComplete）。这样，当相关事件发生时，库会自动调用这些回调函数。
4. 请求级别的回调：除了全局的回调函数，一些函数还允许为单个请求设置成功或失败的回调，这通常通过MQTTAsync_responseOptions结构体来实现。
5. 回调链式编程：应用程序可以围绕这些回调函数构建逻辑，形成一种链式反应式的编程模型，以响应不同的事件。
6. 竞态条件警告：尽管罕见，但理论上存在一种情况，即某个请求的回调（成功或失败）可能在发起该请求的函数返回之前被调用。这种情况下，回调中传递的令牌（token）可能尚未被应用层知晓。这提醒开发者在处理回调时需要考虑潜在的竞态条件问题，请参阅MQTTAsync.c中MQTTAsync _token的Race条件

订阅通配符Subscription wildcards

在MQTT协议中，每个发布的消息都会携带一个主题（topic），用以分类消息内容。MQTT服务器根据这些主题决定哪些订阅者应该接收到发布的消息。理解主题通配符对于灵活设计发布/订阅模式至关重要，尤其是在处理多源数据和未来可能的系统扩展时。

主题层级命名空间

MQTT支持层次化的主题命名空间，通过使用/字符作为层级分隔符，可以组织和管理主题，使之更加结构化和易于维护。例如，SENSOR/1/HUMIDITY这样的主题明确表示了来自编号1的传感器的湿度数据。

通配符

为了提高订阅的灵活性，MQTT定义了两个通配符用于主题过滤：

• #：代表主题层级树中的完整子树，必须作为订阅主题字符串的最后一个字符。例如，SENSOR/#将会匹配所有以SENSOR/开头的主题，包括SENSOR/1/TEMP和SENSOR/2/HUMIDITY。使用#可以订阅某一主题下所有层级的消息。
• +：代表层级结构中的单一层级，使用时位于层级分隔符之间。例如，SENSOR/+/TEMP会匹配SENSOR/1/TEMP和SENSOR/2/TEMP，但不会匹配SENSOR/1/HUMIDITY。+通配符适用于希望订阅特定模式下所有同级主题的情况。

发布者限制

需要注意的是，发布者在发布消息时不能使用通配符来指定主题名称，通配符仅在订阅时有效，用于定义灵活的订阅规则。

系统设计考量

设计一个合理的主题层级结构是系统设计的关键一环，它直接影响到数据的组织、查询效率以及系统的可扩展性。应当考虑当前及未来可能的数据流，确保主题结构既不过于复杂也不过于局限，以适应未来传感器数量的增加或其他变动，同时也要便于管理和理解。正确使用通配符可以大幅提高系统的灵活性和可维护性。

服务质量Quality of service

MQTT协议为客户端和服务器间的消息传输提供了三种服务质量（Quality of Service, QoS）级别，分别是：“最多一次”、“至少一次”和“精确一次”。这些级别确保了消息传递的可靠性和效率，根据应用场景的需求选择合适的QoS等级至关重要。

QoS级别解释：

1. QoS 0: 最多一次（At most once）
   • 消息可能只被传送一次，也可能根本不传送。网络上不进行送达确认，消息也不会被存储。
     如果客户端断开连接或服务器故障，消息可能会丢失。这是最快捷的传输模式，有时被称为“发送即忘”（fire and forget）。
   • 协议不要求服务器必须转发QoS 0级别的消息给离线的客户端，具体行为取决于服务器的实现。
2. QoS 1: 至少一次（At least once）
   • 消息保证至少被传送一次，但在网络异常或确认延迟的情况下可能会重复传送。
     发送方需要本地存储消息，直到收到接收方已发布消息的确认。这样，在需要重发消息时可以使用存储的信息。
   • 相比QoS 0，这提供了更好的可靠性，但牺牲了一定的效率。
3. QoS 2: 精确一次（Exactly once）
   • 消息保证被精确传送一次，是最安全但也是最慢的传输模式。
     同样要求发送方存储消息直到收到确认，但是采用了更为复杂的握手和确认序列来确保消息不会被重复发送。
   • 这个级别确保了即使在网络不稳定或出现错误的情况下，每条消息也只会被接收一次，适合对数据完整性要求极高的场景。

设置QoS

• 发布消息时，通过设置MQTTAsync_message.qos字段来指定消息的QoS级别。
• 订阅消息时，客户端可以通过MQTTAsync_subscribe()或MQTTAsync_subscribeMany()函数设置针对匹配订阅的最高QoS级别。实际转发给订阅者的消息QoS将是这两个值中较低的一个。
  正确选择QoS级别是基于对消息丢失风险、传输速度和系统资源消耗之间的权衡。例如，在需要实时性较高但能容忍少量丢包的场景下，QoS 0可能是最佳选择；而在关键数据传输或财务交易等对消息可靠性要求严格的场景，则应使用QoS 2。

跟踪Tracing

在使用MQTT客户端库时，如果需要对程序运行时的行为进行跟踪（tracing），可以通过环境变量或API调用来控制跟踪功能。这对于调试、性能分析或理解客户端内部运作非常有用。下面是关于如何利用环境变量来配置跟踪功能的说明：

环境变量配置

开启跟踪

• MQTT_C_CLIENT_TRACE：通过设置此环境变量来开启跟踪功能。将其值设为ON或stdout，则跟踪信息会被打印到标准输出（stdout）。如果设置为其他任意值，这个值将被视为文件名，跟踪信息会被写入到该文件中。

控制跟踪详细程度

• MQTT_C_CLIENT_TRACE_LEVEL：此环境变量用于控制跟踪信息的详细程度。有效的值包括：
  • ERROR：仅记录错误信息。
  • PROTOCOL：记录协议相关的详细信息。
  • MINIMUM：记录最少的跟踪信息，通常包含关键步骤和错误。
  • MEDIUM：提供中等程度的详细信息，适合日常调试。
  • MAXIMUM：记录最详尽的信息，包括所有内部操作，适合深入分析。

限制输出行数

• MQTT_C_CLIENT_TRACE_MAX_LINES：此变量用于限制写入文件的跟踪信息行数。默认情况下，最大行数为1000行。当达到最大行数时，如果使用的是文件记录跟踪信息，系统最多会使用两个文件循环覆盖记录新的跟踪条目，即当当前文件满后，最新的跟踪条目会覆盖较早的文件内容。

注意事项

• 确保在启动应用程序之前设置这些环境变量，或者在应用程序代码中动态设置它们（如果支持）。
• 考虑到性能影响，生产环境中通常关闭或仅以最低级别开启跟踪，而在开发和测试阶段可以适当增加跟踪级别以辅助调试。
• 使用文件记录跟踪信息时，注意监控文件系统空间，避免因长时间运行导致磁盘空间不足。

通过这种方式，开发人员可以根据需要调整跟踪的粒度和输出目标，从而更有效地诊断和优化应用程序。

API调用

设置追踪回调函数

• MQTTAsync_traceCallback()：此函数用于设置一个回调函数，每当有追踪信息产生时，这个函数就会被调用。通过这个方法获取的追踪信息与使用环境变量控制追踪时打印的信息相同。这为开发者提供了一个灵活的方式来处理和记录追踪数据。

设置追踪级别

• MQTTAsync_setTraceLevel()：此函数用于设定追踪条目传递给回调函数的最大级别。追踪级别定义了记录的详细程度，可用的级别包括：
  • MQTTASYNC_TRACE_MAXIMUM：选择此级别会捕获并返回所有级别的追踪条目，是最详细的追踪模式。
  • MQTTASYNC_TRACE_MEDIUM：中等追踪级别，通常用于平衡追踪的详细程度和资源消耗。
  • MQTTASYNC_TRACE_MINIMUM：仅记录最少的追踪信息，适用于生产环境或资源受限场景。
  • MQTTASYNC_TRACE_PROTOCOL：专注于协议层面的追踪，有助于分析通信细节。
  • MQTTASYNC_TRACE_ERROR：仅记录错误级别的追踪信息，包括错误、严重错误和致命错误。
  • MQTTASYNC_TRACE_SEVERE：记录严重的错误信息。
  • MQTTASYNC_TRACE_FATAL：仅记录致命错误信息，适用于仅关注最关键问题的情况。

MQTT包跟踪

MQTT数据包追踪是一项非常实用的功能，可以用来打印客户端发送和接收的MQTT数据包。要实现这一功能，可以通过设置以下环境变量来完成：

MQTT_C_CLIENT_TRACE=ON
MQTT_C_CLIENT_TRACE_LEVEL=PROTOCOL
MQTT_C_CLIENT_TRACE_DIR=/你的/追踪/目录路径 # 可选，默认可能会写入默认位置或根据其他追踪设置输出到stdout

您应该看到的输出如下所示：

20130528 155936.813 3 stdout-subscriber -> CONNECT cleansession: 1 (0)
20130528 155936.813 3 stdout-subscriber <- CONNACK rc: 0
20130528 155936.813 3 stdout-subscriber -> SUBSCRIBE msgid: 1 (0)
20130528 155936.813 3 stdout-subscriber <- SUBACK msgid: 1
20130528 155941.818 3 stdout-subscriber -> DISCONNECT (0)

在这个示例中，可以看到以下关键信息：

1. 日期 (date): 表示数据包被发送或接收的具体日期，例如20130528表示2013年5月28日。
2. 时间 (time): 显示数据包处理的时间戳，精确到毫秒，例如155936.813表示当天的15点59分36秒813毫秒。
3. 套接字编号 (socket number): 指示数据通信中所使用的套接字编号，帮助跟踪特定连接的活动，如3。
4. 客户端ID (client id): 表示发出或接收数据包的客户端标识，例如stdout-subscriber。这有助于识别数据包属于哪个客户端实例。
5. 方向 (direction): 用箭头指示数据包的传输方向，->表示数据包从客户端发送到服务器，<-则相反，表示从服务器发送到客户端。
6. 数据包详情 (packet details): 提供了关于数据包类型及其携带的具体信息的详细描述。
   • CONNECT cleansession: 1 (0)：表示一个CONNECT数据包，其中cleansession参数为1，意味着会话清理标志被设置，客户端希望服务器在连接断开后清除其会话状态。
   • CONNACK rc: 0：表示服务器响应的CONNACK数据包，rc: 0意味着连接被成功接受。
   • SUBSCRIBE msgid: 1 (0)：SUBSCRIBE数据包，msgid为1，用于后续确认，订阅请求。
   • SUBACK msgid: 1：服务器对SUBSCRIBE请求的确认，msgid匹配之前的订阅请求。
   • DISCONNECT (0)：客户端发送的DISCONNECT数据包，表示客户端主动断开连接。

这种格式的输出对于理解和诊断MQTT通信过程中的问题非常有帮助，特别是当需要查看协议级别的交互细节时。

默认级别的跟踪

关于重要操作的基本信息，比如连接（connect）、发布（publish）、订阅（subscribe）等操作的基本流程。以下是一个典型的MQTT客户端连接操作在默认追踪级别下的输出示例：

19700101 010000.000 (1152206656) (0)> MQTTClient_connect:893
19700101 010000.000 (1152206656)  (1)> MQTTClient_connectURI:716
20130528 160447.479 Connecting to serverURI localhost:1883
20130528 160447.479 (1152206656)   (2)> MQTTProtocol_connect:98
20130528 160447.479 (1152206656)    (3)> MQTTProtocol_addressPort:48
20130528 160447.479 (1152206656)    (3)< MQTTProtocol_addressPort:73
20130528 160447.479 (1152206656)    (3)> Socket_new:599
20130528 160447.479 New socket 4 for localhost, port 1883
20130528 160447.479 (1152206656)     (4)> Socket_addSocket:163
20130528 160447.479 (1152206656)      (5)> Socket_setnonblocking:73
20130528 160447.479 (1152206656)      (5)< Socket_setnonblocking:78 (0)
20130528 160447.479 (1152206656)     (4)< Socket_addSocket:176 (0)
20130528 160447.479 (1152206656)     (4)> Socket_error:95
20130528 160447.479 (1152206656)     (4)< Socket_error:104 (115)
20130528 160447.479 Connect pending
20130528 160447.479 (1152206656)    (3)< Socket_new:683 (115)
20130528 160447.479 (1152206656)   (2)< MQTTProtocol_connect:131 (115)

在这个示例中，可以看到以下关键信息：

• 日期 (date): 记录了操作发生的日期，例如19700101和20130528。值得注意的是，1970年1月1日可能是日志记录的一个默认或错误日期。
• 时间 (time): 操作发生的具体时间，包括小时、分钟、秒和毫秒，例如010000.000和160447.479。
• 线程ID (thread id): 执行操作的线程标识，例如(1152206656)，这有助于在多线程应用中跟踪特定线程的活动。
• 函数嵌套级别 (function nesting level): 表示函数调用的深度，用括号内的数字表示，如(1), (2), (3)等。数字越大表示调用栈越深，即当前函数被多少层上层函数调用。
• 函数进入或退出标记 (function entry (>) 或 exit (<)): >表示函数被调用的起点，<表示函数执行完毕返回的点。
• 函数名称及源代码行号 (function name : line of source code file): 标识了执行的函数名称及其对应的源代码文件的行号，例如MQTTClient_connect:893。
• 返回值 (return value): 如果有的话，会记录函数执行完毕后的返回值。例如(0)通常表示成功，而(115)可能表示某种错误状态或特定的返回码，需要查阅对应函数的文档或源代码以确定具体含义。

内存分配追踪

内存分配追踪（Memory Allocation Tracing）是一种调试工具，它帮助开发者监控程序中内存的分配与释放情况，尤其是在查找内存泄漏问题时极为有用。
当将追踪级别设置为最大（通常是MAXIMUM或类似级别）时，MQTT客户端库不仅会记录常规的追踪条目，还会详细记录每次内存分配与释放的事件，包括分配/释放的字节数、发生位置（源文件名、行号）以及内存块的地址。消息如下：

20130528 161819.657 Allocating 16 bytes in heap at file /home/icraggs/workspaces/mqrtc/mqttv3c/src/MQTTPacket.c line 177 ptr 0x179f930
20130528 161819.657 Freeing 16 bytes in heap at file /home/icraggs/workspaces/mqrtc/mqttv3c/src/MQTTPacket.c line 201, heap use now 896 bytes

当最后一个MQTT客户端对象被销毁时，如果正在记录跟踪，并且客户端库分配的所有内存尚未释放，则将向跟踪写入一条错误消息。这有助于修复内存泄漏。消息如下所示：

20130528 163909.208 Some memory not freed at shutdown, possible memory leak
20130528 163909.208 Heap scan start, total 880 bytes
20130528 163909.208 Heap element size 32, line 354, file /home/icraggs/workspaces/mqrtc/mqttv3c/src/MQTTPacket.c, ptr 0x260cb00
20130528 163909.208   Content
20130528 163909.209 Heap scan end

这样的追踪记录能够提供以下信息：

• 分配信息：展示每次内存分配的细节，例如“20130528 161819.657 Allocating 16 bytes in heap…”这条记录表明在指定时间、文件及行号处分配了16字节的堆内存，并给出了这块内存的地址。
• 释放信息：记录内存释放的事件，包括释放了多少字节以及释放后堆内存的使用情况，如“Freeing 16 bytes in heap…”。
• 内存泄漏检测：当MQTT客户端对象被销毁且追踪仍在记录时，如果发现有由客户端库分配但未释放的内存，会生成错误消息提示潜在的内存泄漏，如“Some memory not freed at shutdown, possible memory leak”。随后，会列出未释放的内存块详情，包括每个块的大小、分配位置（文件名、行号）及内容摘要，以便开发者定位问题。

这些信息对于开发者来说是极其宝贵的，因为它们直接指向了可能的内存管理问题所在，大大简化了查找和修复内存泄漏的过程。通过分析这些追踪记录，开发者可以确定哪些内存分配没有得到妥善处理，并采取相应措施确保程序的内存使用更加高效和安全。

自动重连Automatic Reconnect

自动重连功能是在客户端库1.1版本中引入的，该功能使得在连接失败的情况下，客户端能够自动尝试重新建立连接。这对于维护系统的稳定性和连续性特别重要。以下是自动重连功能的一些关键点和如何使用它的说明：

开启自动重连

要启用自动重连，你需要在连接选项（connect options）中将automaticReconnect字段设置为非零值。这通常在初始化连接参数时完成。

重连间隔控制

• 最小重试间隔：首次连接失败后，客户端等待的最短时间再尝试重新连接，默认为1秒。
• 最大重试间隔：重试间隔的上限，防止因频繁重试导致的资源过度消耗，默认为60秒。
  当重连尝试失败时，下次尝试的等待时间会翻倍，直到达到最大重试间隔。一旦连接成功建立，重试间隔会被重置回初始值。

连接恢复回调

• 成功重连后，如果之前通过调用MQTTAsync_setConnected设置了MQTTAsync_connected回调函数，那么这个回调函数会被触发。
• 通过这个回调，应用程序可以执行必要的操作，比如重新订阅主题、恢复未完成的消息发布等，以确保服务的连续性。

自动重连的优点

• 简化逻辑：自动重连机制简化了应用层处理网络不稳定情况的逻辑，开发者无需手动编写重连逻辑。
• 提高健壮性：确保客户端在遇到临时网络故障时能自动恢复连接，提高了整体系统的稳定性和可靠性。

实现建议

• 配置合理间隔：根据应用场景调整最小和最大重试间隔，以平衡快速恢复连接的需求和避免对服务器造成不必要的负担。
• 资源管理：在重连成功回调中，确保正确处理之前因断开连接而遗留的问题，如重新初始化资源或状态。
• 异常处理：尽管自动重连提供了便利，仍需在应用中实现连接丢失回调，以处理特殊情况或向用户反馈连接状态变化。
  综上所述，自动重连功能极大地提升了MQTT客户端的鲁棒性和易用性，减少了因网络波动导致的服务中断时间。

离线发布Publish While Disconnected

"在断开连接时发布"这一特性起初并未提供，因为在启用持久化的情况下，消息可能被存储在本地却无法得知是否能成功发送。例如，应用可能因使用了错误的代理（broker）地址或端口而创建客户端。

为了在应用断开连接时也能发布消息，需要使用MQTTAsync_createWithOptions而非MQTTAsync_create来创建客户端对象。在创建选项（::createOptions）中，必须将sendWhileDisconnected字段设置为非零值，并根据需求设置maxBufferedMessages字段的值，其默认值为100，代表客户端最多能缓存多少待发送的消息。

此外，可以使用MQTTAsync_getPendingTokens函数来获取正在等待发送或发送过程尚未完成的消息的标识符（tokens）。这些token可以用来追踪消息的发送状态，以便在重新连接后确认这些消息是否已经成功发送给代理。

总结一下实现要点：

• 使用MQTTAsync_createWithOptions：替代MQTTAsync_create来创建客户端实例，以便能够设置更多高级选项。
• 设置sendWhileDisconnected：在创建选项中将sendWhileDisconnected设为非零，开启断线时发送消息的能力。
• 配置maxBufferedMessages：根据需要设定客户端可以缓冲的最大待发送消息数，默认为100条。
  使用MQTTAsync_getPendingTokens：检查并管理那些因客户端离线而暂存的待发送消息，利用返回的token来追踪消息后续发送的状态。

这一特性的引入，增强了客户端在面对网络不稳定场景时的适应性和消息的可靠性，确保即使暂时与MQTT代理失去连接，也能够继续发送消息并在连接恢复时完成传输。

参考文档

Paho Asynchronous MQTT C Client Library