libwebsockets 实现 MQTT 客户端：minimal-mqtt-client 示例全解析
libwebsockets 实现 MQTT 客户端：minimal-mqtt-client 示例全解析
导读
本文以 libwebsockets 仓库中的 minimal-mqtt-client 示例为主线，完整讲解如何用 libwebsockets 在本地搭建一个可运行的 MQTT 3.1.1 客户端：包括构建、启动 mosquitto broker、订阅/发布消息、观察日志输出，并结合 minimal-mqtt-client.c 源码与 lib/sansio/mqtt 底层实现，剖析连接参数、回调机制、QoS 0/1 发送流程与 keep-alive 心跳等关键原理。读完本文，你将能够独立编译运行该示例，并具备在此基础上开发自定义 MQTT 客户端的基础。
示例定位与项目背景
该示例位于 minimal-examples-lowlevel/mqtt-client/ 目录下，与 minimal-mqtt-client-multi（多连接复用单条 TCP/TLS 连接的演示）并列，属于 libwebsockets 的 MQTT 角色（role）低层示例。它演示的是单一 MQTT 客户端连接的最基础操作：
- 连接本地 broker（localhost）；
- 使用用户名/密码认证；
- 设置 Keep Alive、Clean Start 与遗嘱（Will/LWT）消息；
- 订阅两个主题并发布一长段文本（分别以 QoS0 和 QoS1 发送）；
- 通过回调驱动整个事件循环流程。
libwebsockets 将 MQTT 作为一等网络角色实现，相关协议解析位于 lib/sansio/mqtt/（含 mqtt.c、client/client-mqtt.c），公共 API 头文件为 include/libwebsockets/lws-mqtt.h，示例中所有参数结构体与回调名称均可在其中找到对应定义。
构建示例
进入示例目录后，使用 CMake 构建：
$ cmake . && make
该目录下的 CMakeLists.txt 揭示了编译该示例所需的 libwebsockets 构建选项，任何一项不满足则示例不会被构建：
| 配置项 | 作用 |
|---|---|
LWS_ROLE_MQTT |
启用 MQTT 角色，编译协议栈实现 |
LWS_WITH_CLIENT |
启用客户端连接能力（lws_client_connect_via_info） |
LWS_WITH_SYS_STATE |
启用系统状态机，示例依赖 LWS_SYSTATE_OPERATIONAL 通知来延迟发起连接 |
因此，若你自行编译 libwebsockets 时需要确认这些选项已打开（通常在 CMakeLists.txt 顶层配置中通过 -DLWS_ROLE_MQTT=ON -DLWS_WITH_CLIENT=ON -DLWS_WITH_SYS_STATE=ON 设置）。构建产物为可执行文件 lws-minimal-mqtt-client。
命令行参数与运行
示例支持两个命令行选项：
| 命令行选项 | 含义 |
|---|---|
-d <loglevel> |
十进制调试等级，例如 -d15（数字越大输出越详细） |
-s |
使用 TLS 并连接 8883 端口，替代默认的非加密 1883 端口 |
同时内置 --help 用于打印帮助信息。参数解析由 lws_cmdline_option / lws_cmdline_option_handle_builtin 完成，对应的开关表定义在源码 minimal-mqtt-client.c 中。
前置条件：本地启动 mosquitto broker
$ mosquitto
mosquitto 默认监听 1883 端口（非加密）。若使用 -s 选项，则需要 broker 在 8883 端口提供 TLS 服务，示例目录内已随附测试用证书 mosq-ca.crt、mosq-server.crt 与私钥 mosq-server.key。
运行（非加密模式）
[2020/01/31 10:40:23:7789] U: LWS minimal MQTT client unencrypted [-d<verbosity>][-s]
[2020/01/31 10:40:23:8539] N: lws_mqtt_generate_id: User space provided a client ID 'lwsMqttClient'
[2020/01/31 10:40:23:9893] N: _lws_mqtt_rx_parser: migrated nwsi 0x50febd0 to sid 1 0x5106820
[2020/01/31 10:40:23:9899] U: callback_mqtt: MQTT_CLIENT_ESTABLISHED
[2020/01/31 10:40:23:9967] U: callback_mqtt: WRITEABLE: Subscribing
[2020/01/31 10:40:24:0068] U: callback_mqtt: MQTT_SUBSCRIBED
验证订阅：向测试主题发布消息
mosquitto_pub -h 127.0.0.1 -p 1883 -t test/topic0 -m "hello"
观察客户端收到消息
[2020/01/31 10:40:27:1845] U: callback_mqtt: MQTT_CLIENT_RX
[2020/01/31 10:40:27:1870] N:
[2020/01/31 10:40:27:1945] N: 0000: 74 65 73 74 2F 74 6F 70 69 63 30 test/topic0
[2020/01/31 10:40:27:1952] N:
日志中的 0000: ... 为 lwsl_hexdump_notice 输出的十六进制转储，内容正是收到的主题 test/topic0。程序正常完成一轮订阅/发布后，会输出 Completed: OK 并退出；bad 标志为 1 时则输出 Completed: failed，退出码也会相应为 1（参见 minimal-mqtt-client.c）。
连接参数：一次看懂 CONNECT 报文配置
示例在全局静态结构体 client_connect_param 中声明了连接参数，对应 lws-mqtt.h 中的 lws_mqtt_client_connect_param_t：
static const lws_mqtt_client_connect_param_t client_connect_param = {
.client_id = "lwsMqttClient",
.keep_alive = 60,
.clean_start = 1,
.client_id_nofree = 1,
.username_nofree = 1,
.password_nofree = 1,
.will_param = {
.topic = "good/bye",
.message = "sign-off",
.qos = 0,
.retain = 0,
},
.username = "lwsUser",
.password = "mySecretPassword",
};
各字段含义与底层影响：
- client_id：MQTT 客户端标识。在 client-mqtt.c 的
lws_mqtt_generate_id()中处理：若为空则生成随机 ID（长度取LWS_MQTT_RANDOM_CIDLEN，即 23 字符，符合 MQTT 3.1.1 规范 1～23 字符的要求）；示例提供了固定 ID，故日志显示User space provided a client ID 'lwsMqttClient'。 - keep_alive：Keep Alive 间隔（秒）。底层在 CONNECT 报文中写入该值，并由 retry 策略驱动 PINGREQ。
- clean_start：会话清理标志，对应
LMQCFT_CLEAN_START连接标志（见 lws-mqtt.h）。 - *_nofree 系列：声明 client_id、username、password 字符串的内存由调用方（此处为静态字符串）持有，libwebsockets 不应释放。
- will_param：遗嘱消息（LWT）。设置后底层会置
LMQCFT_WILL_FLAG，并将 QoS 与 retain 标志编码进 CONNECT 标志位（见 client-mqtt.c）。当连接意外中断时，broker 会代客户端向good/bye主题发布sign-off。 - username / password：认证凭据，对应
LMQCFT_USERNAME/LMQCFT_PASSWORD标志。若凭据错误，broker 返回 CONNACK 拒绝，底层lws_mqtt_client_connack_failed()会把原因（如invalid credentials、not authorized、server does not support MQTT protocol 3.1.1等）作为连接失败信息回传给应用（见 client-mqtt.c）。
连接发起：事件驱动与系统状态门控
示例并非在 main() 中立刻连接，而是通过系统状态机延迟到网络就绪后再发起，原因在 system_notify_cb 的注释中写得很清楚：等每个 vhost 的协议初始化完成、网络与时间就绪后再连接，才能正确判断 TLS 证书有效性。回调在状态迁移到 LWS_SYSTATE_OPERATIONAL 时调用 connect_client()。
static int
connect_client(struct lws_context *context)
{
struct lws_client_connect_info i;
memset(&i, 0, sizeof i);
i.mqtt_cp = &client_connect_param;
i.address = "localhost";
i.host = "localhost";
i.protocol = "mqtt";
i.context = context;
i.method = "MQTT";
i.alpn = "mqtt";
i.port = 1883;
if (do_ssl) {
i.ssl_connection = LCCSCF_USE_SSL;
i.ssl_connection |= LCCSCF_ALLOW_SELFSIGNED;
i.port = 8883;
}
if (!lws_client_connect_via_info(&i)) {
lwsl_err("%s: Client Connect Failed\n", __func__);
return 1;
}
return 0;
}
要点：
- protocol / method / alpn 均设为
"mqtt"，这是 libwebsockets 识别 MQTT 角色并走 MQTT 协议栈的关键标识； - 连接目标写死为
localhost:1883，-s时切换为8883并叠加LCCSCF_USE_SSL与LCCSCF_ALLOW_SELFSIGNED（允许自签名证书，配合示例自带测试证书）； main()中设置info.port = CONTEXT_PORT_NO_LISTEN，表示本示例不运行任何服务端监听，纯客户端角色；- 连接信息中同时挂载了 retry 策略：空闲 20 秒后发 PINGREQ，若 25 秒仍无有效活动则判定挂死断开，这是
lws_retry_bo_t结构体在 MQTT keep-alive 上的应用。
事件循环与 MQTT 回调：示例的状态机
示例协议定义将 per_session_data_size 设为 sizeof(struct pss)，struct pss 是一个 4 状态的状态机：
enum {
STATE_SUBSCRIBE, /* subscribe to the topic */
STATE_PUBLISH_QOS0, /* Send the message in QoS0 */
STATE_WAIT_ACK0, /* Wait for the synthetic "ack" */
STATE_PUBLISH_QOS1, /* Send the message in QoS1 */
STATE_WAIT_ACK1, /* Wait for the real ack (or timeout + retry) */
STATE_TEST_FINISH
};
整个测试流程由 LWS_CALLBACK_MQTT_CLIENT_WRITEABLE 回调中的 switch (pss->state) 驱动（minimal-mqtt-client.c），共涉及以下回调（完整枚举见 lws-callbacks.h）：
| 回调 | 触发时机 | 示例中的处理 |
|---|---|---|
LWS_CALLBACK_CLIENT_CONNECTION_ERROR |
连接失败 | 打印错误并退出事件循环 |
LWS_CALLBACK_MQTT_CLIENT_ESTABLISHED |
CONNACK 成功、连接建立 | 请求 WRITEABLE 以发起订阅 |
LWS_CALLBACK_MQTT_CLIENT_WRITEABLE |
可写（含应用主动请求） | 按状态机执行订阅/分块发布 |
LWS_CALLBACK_MQTT_SUBSCRIBED |
SUBACK 收到（或共享连接已订阅） | 请求 WRITEABLE 以开始发布 |
LWS_CALLBACK_MQTT_CLIENT_RX |
收到 PUBLISH 消息 | 转储主题与负载 |
LWS_CALLBACK_MQTT_ACK |
QoS0 发送完本地确认 / QoS1 收到 PUBACK | 推进状态机，完成后退出 |
LWS_CALLBACK_MQTT_RESEND |
QoS1/2 等待 PUBACK/PUBREC 超时 | 重试（最多 3 次） |
LWS_CALLBACK_MQTT_CLIENT_CLOSED |
连接关闭 | 退出事件循环 |
注意回调注释强调：额外的 WRITEABLE 可能不请自来，因此必须通过自身状态机判断是否要利用这次可写机会，这正是示例把业务逻辑挂在 pss->state 上的原因。
订阅流程
在 STATE_SUBSCRIBE 状态下调用：
lws_mqtt_client_send_subcribe(wsi, &sub_param);
sub_param 指定了两个主题（见 minimal-mqtt-client.c）：
static lws_mqtt_topic_elem_t topics[] = {
[0] = { .name = "test/topic0", .qos = QOS0 },
[1] = { .name = "test/topic1", .qos = QOS1 },
};
即分别以 QoS0、QoS1 订阅 test/topic0 与 test/topic1。lws_mqtt_subscribe_param_t 的 num_topics 上限为 LWS_MQTT_MAX_TOPICS（8），lws_mqtt_client_send_subcribe() 返回非零表示失败（lws-mqtt.h）。
分块发布：QoS0 与 QoS1
示例内置了一段 1337 字节的测试文本（TEST_STRING_LEN），发布到 test/topic 主题。MQTT 要求报文在开头即声明完整 PUBLISH 长度，但大帧不必一次写满——lws_mqtt_client_send_publish() 允许分片发送，首片携带头部，后续调用补充剩余负载，末片置 final 标志（API 文档见 lws-mqtt.h）。
示例按 300 字节/片 的块大小循环发送：
pub_param.topic = "test/topic";
pub_param.topic_len = (uint16_t)strlen(pub_param.topic);
pub_param.qos = pss->state == STATE_PUBLISH_QOS0 ? QOS0 : QOS1;
pub_param.payload_len = TEST_STRING_LEN;
chunk = 300;
if (chunk > TEST_STRING_LEN - pss->pos)
chunk = TEST_STRING_LEN - pss->pos;
if (lws_mqtt_client_send_publish(wsi, &pub_param,
test_string + pss->pos, (uint32_t)chunk,
(pss->pos + chunk == TEST_STRING_LEN)))
return -1;
其中最后一个参数表示“这是最后一片”，源码注释也提醒：发送大帧时应按 MTU 量级切块，避免一次性灌入内核缓冲区。
QoS 语义与 ACK / RESEND
- QoS0：没有 PUBACK。底层在最后一片发出后，于事件循环中本地合成
LWS_CALLBACK_MQTT_ACK（即示例状态机中的“合成 ack”）； - QoS1：必须等待 broker 回 PUBACK 才触发
LWS_CALLBACK_MQTT_ACK； - 若等待 PUBACK/PUBREC 超时，底层改发
LWS_CALLBACK_MQTT_RESEND，要求应用按回调携带的 packet ID 重发。示例实现为重试 3 次：if (++pss->retries == 3) lws_default_loop_exit(context);（minimal-mqtt-client.c）。
这也解释了状态机的设计：QoS0 的“ack”是本地合成的（STATE_WAIT_ACK0 几乎立即通过），QoS1 的“ack”是真实的 PUBACK（STATE_WAIT_ACK1 依赖网络往返或超时重发）。LWS_CALLBACK_MQTT_ACK 返回非零可关闭 wsi。
收包处理
LWS_CALLBACK_MQTT_CLIENT_RX 的 in 指向 lws_mqtt_publish_param_t *，其中 topic/topic_len 与 payload/payload_len 描述收到的消息，示例直接以十六进制转储输出：
pub = (lws_mqtt_publish_param_t *)in;
lwsl_hexdump_notice(pub->topic, pub->topic_len);
lwsl_hexdump_notice(pub->payload, pub->payload_len);
底层机制佐证：日志背后的实现
示例运行日志中的两行恰好对应底层两个关键机制，可据此加深理解：
-
lws_mqtt_generate_id: User space provided a client ID 'lwsMqttClient'来自 client-mqtt.c。若应用未提供 client_id，同一函数会用lws_get_random()生成 23 字符随机 ID（字符集为0-9a-zA-Z，共 62 字符）。 -
_lws_mqtt_rx_parser: migrated nwsi ... to sid 1 ...来自 mqtt.c。libwebsockets 的 MQTT 客户端连接存在 nwsi（网络 wsi）与 sid 1 子 wsi 的迁移：CONNECT 阶段由网络连接承载，CONNACK 成功后，连接状态（含客户端 ID、用户空间、解析器）整体移交到新的“sid 1”子连接，随后通过lws_mqtt_set_client_established()进入已建立状态并触发LWS_CALLBACK_MQTT_CLIENT_ESTABLISHED。这一设计与多路复用架构一致——minimal-mqtt-client-multi正是利用该机制让多条独立连接共享同一条 TCP/TLS 链路。 -
Keep Alive 与 PINGREQ 连接参数中的
keep_alive = 60由 retry 策略（minimal-mqtt-client.c）协同驱动：空闲 20 秒发送 PINGREQ，25 秒仍无有效活动则视为挂死。PINGREQ/PINGRESP 的解析与完成处理位于 mqtt.c 及报文类型表（LMQCP_CTOS_PINGREQ/LMQCP_STOC_PINGRESP），保证长时间静默时连接仍能被探测是否存活。
TLS 模式与证书
使用 -s 时，示例以 LCCSCF_USE_SSL | LCCSCF_ALLOW_SELFSIGNED 连接 8883 端口。源码中还有一段条件编译值得注意（minimal-mqtt-client.c）：
#if defined(LWS_WITH_MBEDTLS) || defined(USE_WOLFSSL)
/*
* OpenSSL uses the system trust store. mbedTLS has to be told which
* CA to trust explicitly.
*/
info.client_ssl_ca_filepath = "./mosq-ca.crt";
#endif
即：OpenSSL 后端默认使用系统信任库，而 mbedTLS/WolfSSL 后端必须显式指定 CA 文件，这里指向示例自带的 mosq-ca.crt。使用 TLS 模式前，请确保 broker 的 8883 端口已配置好对应的 mosq-server.crt 与 mosq-server.key。
从示例到实战：如何改造成自己的客户端
基于该示例扩展时，通常只需要修改四处：
- 连接参数：替换
client_connect_param中的 client_id、用户名密码、keep_alive、遗嘱主题，并接入你自己的 broker 地址（修改connect_client()中的i.address/i.port）； - 主题表：按需增删
topics[]与sub_param，注意单次订阅不超过LWS_MQTT_MAX_TOPICS（8 个）； - 发布内容：替换
test_string与TEST_STRING_LEN，并按实际 MTU 调整分块大小（示例为 300 字节/片）； - 回调业务：在
LWS_CALLBACK_MQTT_CLIENT_RX中解析pub->payload做真正的业务处理；pss状态机可按需扩充。
需要理解多连接复用、共享单条 TCP/TLS 传输时，可参考同目录的 minimal-mqtt-client-multi；涉及 QoS2 事务完整流程可参考 minimal-mqtt-client-qos2。完整的 MQTT API 定义（连接/发布/订阅参数、QoS 枚举、控制报文类型、属性与原因码）均在 include/libwebsockets/lws-mqtt.h 中，是进一步开发的权威参考。
小结
通过本文，我们完成了 libwebsockets MQTT 客户端的最小实践闭环：cmake . && make 构建 → 启动 mosquitto → 运行客户端观察 CONNECT/SUBSCRIBE/PUBLISH 日志 → 用 mosquitto_pub 反向验证收包。在此基础上，文中结合 minimal-mqtt-client.c、lws-mqtt.h 与 lib/sansio/mqtt 源码，解释了 client ID 生成、nwsi 到 sid 1 的迁移、QoS0 本地合成 ACK 与 QoS1 真实 PUBACK 的区别、超时重发、keep-alive 心跳以及 mbedTLS 需显式指定 CA 等关键机制，为在 libwebsockets 上开发生产级 MQTT 客户端提供了可复用的起点。