嵌入式工业网关：Modbus转MQTT协议转换代码实现与调试
工业现场大量设备还在跑Modbus RTU（RS485）或Modbus TCP，而上层平台（云、MES、SCADA）普遍走MQTT。嵌入式网关（ARM Cortex-A7/M4 + Linux/RTOS）做协议转换时，最核心的是处理好“定时轮询Modbus从站→打包JSON→发布MQTT”和“订阅MQTT主题→解析→写Modbus寄存器”两条通路。下面以STM32MP157（Cortex-A7跑Linux） + libmodbus + Eclipse Paho MQTT C库为例，讲代码实现和调试要点。
一、整体架构与线程模型
网关开三个线程：
• Modbus轮询线程：每秒循环读取所有从站的寄存器（如温度、压力、转速），存入共享内存结构体。
• MQTT发布线程：每5秒从共享内存取最新数据，拼成JSON字符串，发布到/gw/{device_id}/telemetry。
• MQTT订阅线程：阻塞等待下行消息（如设定值），解析后调用Modbus写寄存器。
// 共享数据结构
typedef struct {
float temperature;
float pressure;
uint16_t rpm;
uint32_t timestamp;
} device_data_t;
device_data_t shared_data;
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
二、Modbus轮询实现
用libmodbus，支持RTU和TCP自动切换（通过URI）。轮询函数：
void *modbus_poll_thread(void *arg) {
modbus_t *ctx = modbus_new_rtu("/dev/ttySTM2", 9600, 'N', 8, 1);
modbus_set_slave(ctx, 1);
modbus_connect(ctx);
uint16_t regs[10];
while (1) {
// 读取起始地址0，长度3个寄存器（温度、压力、转速）
if (modbus_read_registers(ctx, 0, 3, regs) == 3) {
pthread_mutex_lock(&lock);
shared_data.temperature = regs[0] * 0.1f; // 假设精度0.1
shared_data.pressure = regs[1] * 0.01f;
shared_data.rpm = regs[2];
shared_data.timestamp = time(NULL);
pthread_mutex_unlock(&lock);
} else {
fprintf(stderr, "Modbus error: %s\n", modbus_strerror(errno));
}
sleep(1); // 轮询间隔1秒
}
modbus_free(ctx);
return NULL;
}
三、MQTT发布与订阅
使用Paho MQTT C异步客户端。发布线程：
void *mqtt_publish_thread(void *arg) {
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_create(&client, "tcp://broker.emqx.io:1883", "gateway_001",
MQTTCLIENT_PERSISTENCE_NONE, NULL);
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
MQTTClient_connect(client, &conn_opts);
char topic[64], payload[256];
snprintf(topic, sizeof(topic), "/gw/%s/telemetry", "gw_001");
while (1) {
pthread_mutex_lock(&lock);
snprintf(payload, sizeof(payload),
"{ \"temp\":%.1f,\"press\":%.2f,\"rpm\":%u,\"ts\":%lu }",
shared_data.temperature, shared_data.pressure,
shared_data.rpm, shared_data.timestamp);
pthread_mutex_unlock(&lock);
MQTTClient_publish(client, topic, strlen(payload), payload, 0, 0, NULL);
sleep(5);
}
MQTTClient_disconnect(client, 10000);
MQTTClient_destroy(&client);
return NULL;
}
订阅线程需要注册回调，收到/gw/gw_001/command后解析JSON并写Modbus：
int msg_arrived(void *context, char *topicName, int topicLen, MQTTClient_message *msg) {
// 解析JSON，提取setpoint字段，调用modbus_write_register
cJSON *root = cJSON_Parse(msg->payload);
if (root) {
cJSON *sp = cJSON_GetObjectItem(root, "setpoint");
if (sp) {
uint16_t val = (uint16_t)(sp->valuedouble * 10);
modbus_write_register(modbus_ctx, 10, val); // 写入地址10
}
cJSON_Delete(root);
}
MQTTClient_freeMessage(&msg);
MQTTClient_free(topicName);
return 1;
}
四、调试要点
1. Modbus抓包：用mbpoll命令行工具或串口助手验证从站响应。如果modbus_read_registers返回-1，检查波特率、校验、从站地址。
2. MQTT调试：用Mosquitto客户端订阅/gw/#看发布内容；用mosquitto_pub发下行命令测试订阅。
3. JSON解析：推荐cJSON库，注意浮点数精度，避免字符串溢出。
4. 日志输出：在关键点加syslog或文件日志，便于定位“Modbus通了但MQTT没发”的问题。
五、三个翻车高发点
• Modbus超时设置：默认1秒，如果从站响应慢（如老PLC），需modbus_set_response_timeout(ctx, 2, 0)，否则频繁超时。
• MQTT QoS：工业场景建议QoS=1（至少一次），并开启持久会话，防止断线丢消息。
• JSON键名与云端对齐：云端期望temperature，代码里写成temp，数据上去了但平台不认。提前约定好字段名。
下载文档
