概念先知道
- MQTT：物联网最常用的消息协议，采用“发布/订阅（Pub/Sub）”模型：设备把消息发到主题（Topic），关心该主题的设备收到消息。发布者和订阅者互不认识，通过**Broker（服务器）**中转。
- 主题（Topic）：消息的“频道名”，如
mqtt_test、device/1/temperature，用/分层。 - Broker：MQTT 服务器，负责接收发布、转发给订阅者。公共测试服务器有
test.mosquitto.org、EMQX 在线服务等。 - QoS：消息服务质量等级。例程用
MQTT_PUBLISH_QOS_0（最多一次，最快但可能丢）。 - Client（客户端）：例程用 paho 风格接口
mqtt_init / mqtt_connect / mqtt_publish，配合mqtt_sync周期性驱动收发。
例程功能简介
本页对应博流官方 SDK 的 wifi_mqtt 例程（examples/wifi/sta/wifi_mqtt），其中发布端（wifi_mqtt_pub.c）：
mqtt_pub <ip> <port>命令连接 Broker（默认主题mqtt_test）；mqtt_init建立客户端（发送缓冲 2048 字节、接收缓冲 1024 字节）；mqtt_connect发起连接（匿名会话 + 干净会话标志），失败打印mqtt_error_str；- 每 3 秒
mqtt_publish一次{"hello mqtt !"}，client_refresher任务每 100ms 调用mqtt_sync驱动收发。 - 同族例程：同目录提供订阅端
mqtt_sub <ip> <port>（wifi_mqtt_sub.c）、TLS 加密版mqtts_pub/mqtts_sub（见 MQTTS 页）。
操作步骤
进入例程目录
在终端进入 SDK 的 MQTT 例程目录（前提：环境已按快速开始（Linux）或Windows搭好）：
cd examples/wifi/sta/wifi_mqtt编译工程
统一填写 bl616：
make CHIP=bl616 BOARD=bl616dk烧录固件
按住 BOOT 键短按 EN/RST 进入下载模式后烧录：
make flash CHIP=bl616 COMX=/dev/ttyUSB0准备 MQTT Broker
准备一个可用的 MQTT 服务器：可用公共测试服务器（如 test.mosquitto.org:1883），也可在电脑上自建（mosquitto -p 1883）。先用订阅端确认服务器正常，例如：
mosquitto_sub -h 127.0.0.1 -p 1883 -t mqtt_test连接 Wi-Fi 并发布消息
串口助手波特率 2000000，连接路由器后执行 mqtt_pub <服务器IP> <端口>：
wifi_sta_connect Your_SSID 12345678
mqtt_pub 192.168.1.143 1883运行验证
模组每 3 秒向主题 mqtt_test 发布一条 {"hello mqtt !"} 消息，电脑上的 mosquitto_sub 应同步收到；Ctrl+C 退出并打印停止信息。
代码执行流程
MQTT 发布端从启动到发布消息的完整流程如下：
例程调用的 API 介绍
mqtt_init(&client, &handle, sendbuf, len, recvbuf, len, cb)
初始化 MQTT 客户端，绑定 socket 句柄与收发缓冲区。
参数：
client：struct mqtt_clienthandle：struct custom_socket_handle（TCP 类型 + fd）sendbuf/recvbuf：发送/接收缓冲区（例程 2048 / 1024 字节）cb：收到 PUBLISH 消息时的回调
返回值：无
mqtt_connect(&client, client_id, ...)
向 Broker 发起连接。例程传 client_id = NULL（匿名）+ MQTT_CONNECT_CLEAN_SESSION，超时 400ms。
参数：
client：客户端对象client_id：客户端 ID，可为NULLconnect_flags：连接标志
返回值：MQTT_OK 成功；其他错误码
mqtt_publish(&client, topic, msg, len, QoS)
向指定主题发布一条消息，例程主题 mqtt_test、QoS 0。
参数：
client：客户端对象topic：主题字符串msg/len：消息内容与长度QoS：MQTT_PUBLISH_QOS_0等
返回值：MQTT_OK 成功
完整代码
以下为 wifi_mqtt_pub.c 完整源码，与官方示例（examples/wifi/sta/wifi_mqtt）一致，默认折叠，点击展开：
📜 点击展开 wifi_mqtt/main.c 完整代码
c
/****************************************************************************
*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership. The
* ASF licenses this file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance with the
* License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*
****************************************************************************/
/****************************************************************************
* Included Files
****************************************************************************/
#include "FreeRTOS.h"
#include "task.h"
#include "timers.h"
#include <lwip/tcpip.h>
#include <lwip/sockets.h>
#include <lwip/netdb.h>
#include "wifi_mgmr_ext.h"
#include "bflb_irq.h"
#include "bflb_uart.h"
#include "rfparam_adapter.h"
#include "board.h"
#include "shell.h"
#ifndef BL602
#include "fhost_api.h"
#include "wifi_mgmr.h"
#endif
#include "async_event.h"
#include "mm.h"
#define DBG_TAG "MAIN"
#include "log.h"
struct bflb_device_s *gpio;
/****************************************************************************
* Pre-processor Definitions
****************************************************************************/
#define WIFI_STACK_SIZE (1536)
#define TASK_PRIORITY_FW (16)
/****************************************************************************
* Private Types
****************************************************************************/
/****************************************************************************
* Private Data
****************************************************************************/
static struct bflb_device_s *uart0;
static TaskHandle_t wifi_fw_task __attribute__((unused));
extern void shell_init_with_task(struct bflb_device_s *shell);
extern void wifi_event_handler(async_input_event_t ev, void *priv);
#ifdef BL602
extern void wifi_task_create(void);
extern int fhost_init(void);
extern int wifi_mgmr_task_start(void);
#endif
/****************************************************************************
* Private Function Prototypes
****************************************************************************/
/****************************************************************************
* Functions
****************************************************************************/
void wifi_start_firmware_task(void *param)
{
LOG_I("Starting wifi ...\r\n");
async_register_event_filter(EV_WIFI, wifi_event_handler, NULL);
wifi_task_create();
LOG_I("Starting fhost ...\r\n");
fhost_init();
vTaskDelete(NULL);
}
volatile uint32_t wifi_state = 0;
void wifi_event_handler(async_input_event_t ev, void *priv)
{
uint32_t code = ev->code;
switch (code) {
case CODE_WIFI_ON_INIT_DONE: {
LOG_I("[APP] [EVT] %s, CODE_WIFI_ON_INIT_DONE\r\n", __func__);
wifi_mgmr_task_start();
} break;
case CODE_WIFI_ON_MGMR_DONE: {
LOG_I("[APP] [EVT] %s, CODE_WIFI_ON_MGMR_DONE\r\n", __func__);
} break;
case CODE_WIFI_ON_SCAN_DONE: {
LOG_I("[APP] [EVT] %s, CODE_WIFI_ON_SCAN_DONE\r\n", __func__);
wifi_mgmr_sta_scanlist();
} break;
case CODE_WIFI_ON_CONNECTED: {
LOG_I("[APP] [EVT] %s, CODE_WIFI_ON_CONNECTED\r\n", __func__);
void mm_sec_keydump();
mm_sec_keydump();
} break;
case CODE_WIFI_ON_GOT_IP: {
wifi_state = 1;
LOG_I("[APP] [EVT] %s, CODE_WIFI_ON_GOT_IP\r\n", __func__);
LOG_I("[SYS] Memory left is %d Bytes\r\n", kfree_size(0));
} break;
case CODE_WIFI_ON_DISCONNECT: {
wifi_state = 0;
LOG_I("[APP] [EVT] %s, CODE_WIFI_ON_DISCONNECT\r\n", __func__);
} break;
case CODE_WIFI_ON_AP_STARTED: {
LOG_I("[APP] [EVT] %s, CODE_WIFI_ON_AP_STARTED\r\n", __func__);
} break;
case CODE_WIFI_ON_AP_STOPPED: {
LOG_I("[APP] [EVT] %s, CODE_WIFI_ON_AP_STOPPED\r\n", __func__);
} break;
case CODE_WIFI_ON_AP_STA_ADD: {
LOG_I("[APP] [EVT] [AP] [ADD] %lld\r\n", xTaskGetTickCount());
} break;
case CODE_WIFI_ON_AP_STA_DEL: {
LOG_I("[APP] [EVT] [AP] [DEL] %lld\r\n", xTaskGetTickCount());
} break;
default: {
LOG_I("[APP] [EVT] Unknown code %u \r\n", code);
}
}
}
int main(void)
{
board_init();
uart0 = bflb_device_get_by_name("uart0");
shell_init_with_task(uart0);
if (0 != rfparam_init(0, NULL, 0)) {
LOG_I("PHY RF init failed!\r\n");
return 0;
}
LOG_I("PHY RF init success!\r\n");
tcpip_init(NULL, NULL);
xTaskCreate(wifi_start_firmware_task, "wifi init", 1024, NULL, 10, NULL);
vTaskStartScheduler();
while (1) {
}
}📜 点击展开 wifi_mqtt_pub.c 完整代码
c
#include "FreeRTOS_POSIX.h"
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <lwip/errno.h>
#include <netdb.h>
#include "utils_getopt.h"
#include "mqtt.h"
#include "shell.h"
static uint8_t sendbuf[2048]; /* sendbuf should be large enough to hold multiple whole mqtt messages */
static uint8_t recvbuf[1024]; /* recvbuf should be large enough any whole mqtt message expected to be received */
static shell_sig_func_ptr abort_exec;
static TaskHandle_t client_daemon;
static int test_sockfd;
static const char* addr;
/*
A template for opening a non-blocking POSIX socket.
*/
static int open_nb_socket(const char* addr, const char* port);
static int open_nb_socket(const char* addr, const char* port) {
struct addrinfo hints = {0};
hints.ai_family = AF_UNSPEC; /* IPv4 or IPv6 */
hints.ai_socktype = SOCK_STREAM; /* Must be TCP */
int sockfd = -1;
int rv;
struct addrinfo *p, *servinfo;
/* get address information */
rv = getaddrinfo(addr, port, &hints, &servinfo);
if(rv != 0) {
printf("Failed to open socket (getaddrinfo): %s\r\n", rv);
return -1;
}
/* open the first possible socket */
for(p = servinfo; p != NULL; p = p->ai_next) {
sockfd = socket(p->ai_family, p->ai_socktype, p->ai_protocol);
if (sockfd == -1) continue;
/* connect to server */
rv = connect(sockfd, p->ai_addr, p->ai_addrlen);
if(rv == -1) {
close(sockfd);
sockfd = -1;
continue;
}
break;
}
/* free servinfo */
freeaddrinfo(servinfo);
/* make non-blocking */
if (sockfd != -1) {
int iMode = 1;
ioctlsocket(sockfd, FIONBIO, &iMode);
}
return sockfd;
}
/**
* @brief The function will be called whenever a PUBLISH message is received.
*/
static void publish_callback_1(void** unused, struct mqtt_response_publish *published);
/**
* @brief The client's refresher. This function triggers back-end routines to
* handle ingress/egress traffic to the broker.
*
* @note All this function needs to do is call \ref __mqtt_recv and
* \ref __mqtt_send every so often. I've picked 100 ms meaning that
* client ingress/egress traffic will be handled every 100 ms.
*/
static void client_refresher(void* client);
/**
* @brief Safelty closes the \p sockfd and cancels the \p client_daemon before \c exit.
*/
static void test_close(int sig)
{
if (test_sockfd)
{
close(test_sockfd);
}
printf("mqtt_pub stop publish to %s\r\n", addr);
abort_exec(sig);
vTaskDelete(client_daemon);
}
static int example_mqtt(int argc, const char *argv[])
{
const char* port = NULL;
const char* topic;
int ret = 0;
// int argc = 0;
abort_exec = shell_signal(1, test_close);
/* get address (argv[1] if present) */
if (argc > 1) {
addr = argv[1];
}
/* get port number (argv[2] if present) */
if (argc > 2) {
port = argv[2];
}
/* get the topic name to publish */
topic = "mqtt_test";
/* open the non-blocking TCP socket (connecting to the broker) */
test_sockfd = open_nb_socket(addr, port);
struct custom_socket_handle handle;
handle.type = MQTTC_PAL_CONNTION_TYPE_TCP;
handle.ctx.fd = test_sockfd;
if (test_sockfd < 0) {
printf("Failed to open socket: %d\r\n", test_sockfd);
test_close(SHELL_SIGINT);
}
/* setup a client */
struct mqtt_client client;
mqtt_init(&client, &handle, sendbuf, sizeof(sendbuf), recvbuf, sizeof(recvbuf), publish_callback_1);
/* Create an anonymous session */
const char* client_id = NULL;
/* Ensure we have a clean session */
uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
/* Send connection request to the broker. */
ret = mqtt_connect(&client, client_id, NULL, NULL, 0, NULL, NULL, connect_flags, 400);
if (ret != MQTT_OK)
{
printf("fail \r\n");
}
/* check that we don't have any errors */
if (client.error != MQTT_OK) {
printf("error: %s\r\n", mqtt_error_str(client.error));
test_close(SHELL_SIGINT);
}
/* start a thread to refresh the client (handle egress and ingree client traffic) */
xTaskCreate(client_refresher, (char*)"client_ref", 1024, &client, 10, &client_daemon);
printf("Press ENTER to publish hello.\r\n");
printf("Press CTRL-C to exit.\r\n");
/* block wait CTRL-C exit */
while(1) {
/* print a message */
char application_message[256] = {"{\"hello mqtt !\"}\r\n"};
printf("%s published : \"%s\"\r\n", argv[0], application_message);
mqtt_publish(&client, topic, application_message, strlen(application_message) + 1, MQTT_PUBLISH_QOS_0);
if (client.error != MQTT_OK) {
printf("error: %s\r\n", mqtt_error_str(client.error));
test_close(SHELL_SIGINT);
}
vTaskDelay(3000);
}
/* disconnect */
/* exit */
test_close(SHELL_SIGINT);
return 0;
}
static void publish_callback_1(void** unused, struct mqtt_response_publish *published)
{
/* not used in this example */
}
static void client_refresher(void* client)
{
while(1)
{
mqtt_sync((struct mqtt_client*) client);
vTaskDelay(100);
}
}
#ifdef CONFIG_SHELL
#include <shell.h>
extern uint32_t wifi_state;
static int check_wifi_state(void)
{
if (wifi_state == 1)
{
return 0;
} else {
return 1;
}
}
static int cmd_mqtt_publisher(int argc, const char **argv)
{
uint32_t ret = 0;
ret = check_wifi_state();
if (ret != 0) {
printf("your wifi not connected!\r\n");
return 0;
}
// xTaskCreate(example_mqtt,(char*)"test_mqtt", 8192, argv, 10, NULL);
example_mqtt(argc, argv);
return 0;
}
SHELL_CMD_EXPORT_ALIAS(cmd_mqtt_publisher, mqtt_pub, mqtt publisher);
#endif📜 点击展开 wifi_mqtt/wifi_mqtt_sub.c 完整代码
c
/**
* @file
* A simple program that subscribes to a topic.
*/
#include "FreeRTOS_POSIX.h"
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <lwip/errno.h>
#include <netdb.h>
#include "utils_getopt.h"
#include "mqtt.h"
#include "shell.h"
static uint8_t sendbuf[2048]; /* sendbuf should be large enough to hold multiple whole mqtt messages */
static uint8_t recvbuf[1024]; /* recvbuf should be large enough any whole mqtt message expected to be received */
static shell_sig_func_ptr abort_exec;
static TaskHandle_t client_daemon;
static int test_sockfd;
static const char* addr;
/*
A template for opening a non-blocking POSIX socket.
*/
static int open_nb_socket(const char* addr, const char* port);
static int open_nb_socket(const char* addr, const char* port) {
struct addrinfo hints = {0};
hints.ai_family = AF_UNSPEC; /* IPv4 or IPv6 */
hints.ai_socktype = SOCK_STREAM; /* Must be TCP */
int sockfd = -1;
int rv;
struct addrinfo *p, *servinfo;
/* get address information */
rv = getaddrinfo(addr, port, &hints, &servinfo);
if(rv != 0) {
printf("Failed to open socket (getaddrinfo): %s\r\n", rv);
return -1;
}
/* open the first possible socket */
for(p = servinfo; p != NULL; p = p->ai_next) {
sockfd = socket(p->ai_family, p->ai_socktype, p->ai_protocol);
if (sockfd == -1) continue;
/* connect to server */
rv = connect(sockfd, p->ai_addr, p->ai_addrlen);
if(rv == -1) {
close(sockfd);
sockfd = -1;
continue;
}
break;
}
/* free servinfo */
freeaddrinfo(servinfo);
/* make non-blocking */
if (sockfd != -1) {
int iMode = 1;
ioctlsocket(sockfd, FIONBIO, &iMode);
}
return sockfd;
}
/**
* @brief The function will be called whenever a PUBLISH message is received.
*/
static void publish_callback_1(void** unused, struct mqtt_response_publish *published);
/**
* @brief The client's refresher. This function triggers back-end routines to
* handle ingress/egress traffic to the broker.
*
* @note All this function needs to do is call \ref __mqtt_recv and
* \ref __mqtt_send every so often. I've picked 100 ms meaning that
* client ingress/egress traffic will be handled every 100 ms.
*/
static void client_refresher(void* client);
/**
* @brief Safelty closes the \p sockfd and cancels the \p client_daemon before \c exit.
*/
static void test_close(int sig)
{
if (test_sockfd)
{
close(test_sockfd);
}
printf("mqtt_sub disconnecting from %s\r\n", addr);
abort_exec(sig);
vTaskDelete(client_daemon);
}
static int example_mqtt(int argc, const char *argv[])
{
const char* port = NULL;
const char* topic;
int ret = 0;
// int argc = 0;
abort_exec = shell_signal(1, test_close);
/* get address (argv[1] if present) */
if (argc > 1) {
addr = argv[1];
}
/* get port number (argv[2] if present) */
if (argc > 2) {
port = argv[2];
}
/* get the topic name to publish */
topic = "mqtt_test";
/* open the non-blocking TCP socket (connecting to the broker) */
test_sockfd = open_nb_socket(addr, port);
struct custom_socket_handle handle;
handle.type = MQTTC_PAL_CONNTION_TYPE_TCP;
handle.ctx.fd = test_sockfd;
if (test_sockfd < 0) {
printf("Failed to open socket: %d\r\n", test_sockfd);
test_close(SHELL_SIGINT);
}
/* setup a client */
struct mqtt_client client;
mqtt_init(&client, &handle, sendbuf, sizeof(sendbuf), recvbuf, sizeof(recvbuf), publish_callback_1);
/* Create an anonymous session */
const char* client_id = NULL;
/* Ensure we have a clean session */
uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
/* Send connection request to the broker. */
ret = mqtt_connect(&client, client_id, NULL, NULL, 0, NULL, NULL, connect_flags, 400);
if (ret != MQTT_OK)
{
printf("fail \r\n");
}
/* check that we don't have any errors */
if (client.error != MQTT_OK) {
printf("error: %s\r\n", mqtt_error_str(client.error));
test_close(SHELL_SIGINT);
}
/* start a thread to refresh the client (handle egress and ingree client traffic) */
xTaskCreate(client_refresher, (char*)"client_ref", 1024, &client, 10, &client_daemon);
/* subscribe */
mqtt_subscribe(&client, topic, 0);
printf("%s listening for '%s' messages.\r\n", argv[0], topic);
printf("Press CTRL-C to exit.\r\n");
/* block wait CTRL-C exit */
while(1) {
vTaskDelay(100);
}
/* disconnect */
/* exit */
test_close(SHELL_SIGINT);
return 0;
}
static void publish_callback_1(void** unused, struct mqtt_response_publish *published)
{
/* note that published->topic_name is NOT null-terminated (here we'll change it to a c-string) */
char* topic_name = (char*) malloc(published->topic_name_size + 1);
char* topic_msg = (char*) malloc(published->application_message_size + 1);
if (topic_name && topic_msg) {
memcpy(topic_name, published->topic_name, published->topic_name_size);
topic_name[published->topic_name_size] = '\0';
memcpy(topic_msg, published->application_message, published->application_message_size);
topic_msg[published->application_message_size] = '\0';
printf("Received publish('%s'): %s\r\n", topic_name, topic_msg);
} else {
printf("No memory to receive published msg\r\n");
}
if(topic_name) {
free(topic_name);
}
if(topic_msg) {
free(topic_msg);
}
}
static void client_refresher(void* client)
{
while(1)
{
mqtt_sync((struct mqtt_client*) client);
vTaskDelay(100);
}
}
#ifdef CONFIG_SHELL
#include <shell.h>
extern uint32_t wifi_state;
static int check_wifi_state(void)
{
if (wifi_state == 1)
{
return 0;
} else {
return 1;
}
}
static int cmd_mqtt_subscribe(int argc, const char **argv)
{
uint32_t ret = 0;
ret = check_wifi_state();
if (ret != 0) {
printf("your wifi not connected!\r\n");
return 0;
}
// xTaskCreate(example_mqtt,(char*)"test_mqtt", 8192, argv, 10, NULL);
example_mqtt(argc, argv);
return 0;
}
SHELL_CMD_EXPORT_ALIAS(cmd_mqtt_subscribe, mqtt_sub, mqtt subscribe);
#endifFAQ
mqtt_pub 提示 your wifi not connected
例程只有在 wifi_state == 1（CODE_WIFI_ON_GOT_IP）时才允许执行命令；先执行 wifi_sta_connect 并确认日志出现 CODE_WIFI_ON_GOT_IP。
连接公共 Broker 失败
公共测试服务器（test.mosquitto.org 等）可能不稳定或被网络限制；建议在电脑上自建 mosquitto 服务，用模组直连局域网 IP 调试。匿名连接时确保 Broker 允许匿名（默认允许）。
遇到问题？
如有其他问题，请到统一的提问与讨论区：Ai-Thinker Discussions