linux mqtt客户端

大家好，又见面了，我是你们的朋友全栈君。如果您正在找激活码,请点击查看最新教程,关注关注公众号 “全栈程序员社区” 获取激活教程,可能之前旧版本教程已经失效.最新Idea2022.1教程亲测有效,一键激活。

Jetbrains全家桶1年46，售后保障稳定

实现功能：

（1）定时30s发送心跳包；

（2）接收 mqtt 数据包，解析函数是 user_recv_handle_cb;

（3）定时 PERIOD_TIME 发布自身订阅的主题信息，即循环 PERIOD_TIME 发啥收啥。

说明：

（1）主要根据庆科的MiCO_A_v3.2.0/demos/net/mqtt_client 的 stm32 freeRTOS 移植到 linux 平台。

（2）实现方式：select、queue 、pthread。

核心源码：

/*************************************** 描述***********************
作者： lee
日期： 2019/7/2
文件名：mqtt_client.c
功能描述：
1.定时30s发送心跳包
2.接收 mqtt 数据包，解析函数是user_recv_handle_cb
3.定时  PERIOD_TIME   发布 自身订阅的主题 信息，即循环 PERIOD_TIME 发啥收啥
**********************************************************************/
#include "./libraries/protocols/mqtt/MQTTClient.h"
#include "/usr/local/include/uv.h"
#include "pthread.h"
#include "sys/select.h"
#include "sys/queue.h"
/*********************************
*              Macros
***********************************************/
#define app_log(M, ...) custom_log("APP", M, ##__VA_ARGS__)
#define mqtt_log(M, ...) custom_log("MQTT", M, ##__VA_ARGS__)
#define MQTT_CMD_TIMEOUT 5000 // 5s
#define MAX_MQTT_TOPIC_SIZE  (256)
#define MAX_MQTT_DATA_SIZE   (1024)
#define MQTT_SERVER "127.0.0.1"
//#define MQTT_SERVER "test.mosquitto.org"
#define MQTT_SERVER_PORT 1883
#define PERIOD_TIME   2000  // 2s
/***********************************************
*              Constants
***********************************************/
#define MQTT_CLIENT_ID  "MiCO_MQTT_Client"
#define MQTT_CLIENT_USERNAME NULL
#define MQTT_CLIENT_PASSWORD NULL
#define MQTT_CLIENT_KEEPALIVE 30
#define MQTT_CLIENT_SUB_TOPIC "mico/test/send" // loop msg
#define MQTT_CLIENT_PUB_TOPIC "mico/test/send"
#define MQTT_YIELD_TMIE 5000 // 5s
#define MQTT_CLIENT_PUB_MSG "mico_mqtt_client_test_data_1234567890"
/***********************************************
*              Structures
***********************************************/
typedef struct {   
char topic[MAX_MQTT_TOPIC_SIZE];
char qos;
char retained;
uint8_t data[MAX_MQTT_DATA_SIZE];
uint32_t datalen;
} s_MQTT_Data_Packet_Info;
struct node{
STAILQ_ENTRY(node) next;
void (*fp) (void*);  
void *data;
};
/***********************************************
*              Function Declarations
***********************************************/
void user_send_cb(void* data);
/***********************************************
*              Variables Definitions
***********************************************/
uv_req_t mqtt_client_recv_handle, mqtt_client_send_handle;
volatile static bool no_mqtt_msg_exchange = true;
Client c; // mqtt client object
Network n; // socket network for mqtt client
STAILQ_HEAD(head, node);
struct head *lhead = 0;
void user_recv_handle_cb(void* data){
s_MQTT_Data_Packet_Info *p_recv_msg = (s_MQTT_Data_Packet_Info *)data;
if (p_recv_msg)
{
实际工程中替换
app_log("\t\t\t\t\tuser get data success! from_topic=[%s], msg=[%ld][%s].\r\n", p_recv_msg->topic, p_recv_msg->datalen, p_recv_msg->data);  
free(p_recv_msg);
p_recv_msg = NULL;
}          
}
// call back, msg received from mqtt server
static void messageArrived(MessageData* md){
s_MQTT_Data_Packet_Info* p_recv_msg = NULL; 
MQTTMessage* pMessage = md->message;
p_recv_msg = (s_MQTT_Data_Packet_Info *)calloc(1, sizeof(s_MQTT_Data_Packet_Info));
if (p_recv_msg == NULL)  
{
mqtt_log("malloc内存分配不足");
return;
}
p_recv_msg->datalen = pMessage->payloadlen;
p_recv_msg->qos = pMessage->qos;
p_recv_msg->retained = pMessage->retained;
strncpy(p_recv_msg->topic, md->topicName->lenstring.data, md->topicName->lenstring.len);
memcpy(p_recv_msg->data, pMessage->payload, p_recv_msg->datalen + 1);  // lee:  !!!!!!!!!!!!!!!!!!!!   p_recv_msg->datalen + 1  不加1，会出现段错误 
// mqtt_client_recv_handle.data = &p_recv_msg;
// uv_queue_work(loop, &mqtt_client_recv_handle, user_recv_handle_cb, NULL);    //  lee !!!!!!!!!!!!!!!!!!!! 发现  libuv中的工作队列不能和select混用
struct node process_func;
process_func.data = p_recv_msg;
process_func.fp = user_recv_handle_cb;
STAILQ_INSERT_TAIL(lhead, &process_func, next); 
p_recv_msg = NULL; 
}
static OSStatus mqtt_client_release(Client *c, Network *n){
OSStatus err = kNoErr;
if (c->isconnected) MQTTDisconnect(c);
n->disconnect(n);// close connection
if (MQTT_SUCCESS != MQTTClientDeinit(c)){
app_log("MQTTClientDeinit failed!");
err = kDeletedErr;
}
return err;
}
void* work_thread(void* arg){
s_MQTT_Data_Packet_Info* p_send_msg = NULL;
Timer period_timer;
InitTimer(&period_timer);
countdown_ms(&period_timer, PERIOD_TIME);
while (1)
{
if (!STAILQ_EMPTY(lhead)){
struct node* pfirst_node = STAILQ_FIRST(lhead);
pfirst_node->fp(pfirst_node->data);
STAILQ_REMOVE_HEAD(lhead, next);
} 
while (expired(&period_timer))
{
if (c.isconnected){
p_send_msg = (s_MQTT_Data_Packet_Info*) calloc(1, sizeof(s_MQTT_Data_Packet_Info));                
if (p_send_msg == NULL) {
mqtt_log("没有内存可用");
continue;
}
p_send_msg->qos = 0;
p_send_msg->retained = 0;
p_send_msg->datalen = strlen(MQTT_CLIENT_PUB_MSG);
memcpy(p_send_msg->data, MQTT_CLIENT_PUB_MSG, p_send_msg->datalen);
strncpy(p_send_msg->topic, MQTT_CLIENT_PUB_TOPIC, MAX_MQTT_TOPIC_SIZE);
struct node process_func;
process_func.data = p_send_msg;
process_func.fp = user_send_cb;
STAILQ_INSERT_TAIL(lhead, &process_func, next); 
p_send_msg = NULL;
}
else
{
mqtt_log("MQTT client does not init ok");
}
countdown_ms(&period_timer, PERIOD_TIME);   
} 
}
pthread_exit(NULL);   
}
void* mqtt_client_thread(void* arg){
OSStatus err = kUnknownErr;
int rc = -1;
fd_set readfds;
struct timeval t = { 0, MQTT_YIELD_TMIE * 1000};
ssl_opts ssl_settings;
MQTTPacket_connectData connectData = MQTTPacket_connectData_initializer;
memset(&c, 0, sizeof(c));
memset(&n, 0, sizeof(n));
MQTT_start:
// create network connection
ssl_settings.ssl_enable = false;
mqtt_log("enter into mqtt client thread.");
while (1){
rc = NewNetwork(&n, MQTT_SERVER, MQTT_SERVER_PORT, ssl_settings);
if (rc == MQTT_SUCCESS) break;
mqtt_log("ERROR: MQTT network connection err=%d, reconnect after 3s...", rc);
sleep(3);
}
mqtt_log("MQTT network connection success!");
// 2.init mqtt client
rc = MQTTClientInit(&c, &n, MQTT_CMD_TIMEOUT);
require_noerr_string(rc, MQTT_reconnect, "ERROR: MQTT client init err.");
mqtt_log("MQTT client init success!");
// 3.create mqtt client connection
connectData.willFlag = 0;
connectData.MQTTVersion = 4; // 3: 3.1, 4: v3.1.1
connectData.clientID.cstring = MQTT_CLIENT_ID;
connectData.username.cstring = MQTT_CLIENT_USERNAME;
connectData.password.cstring = MQTT_CLIENT_PASSWORD;
connectData.keepAliveInterval = MQTT_CLIENT_KEEPALIVE;
connectData.cleansession = 1;
rc = MQTTConnect(&c, &connectData);
require_noerr_string(rc, MQTT_reconnect, "ERROR: MQTT client connect err.");
mqtt_log("MQTT client connect success!");
// 4.mqtt client subscribe
rc = MQTTSubscribe(&c, MQTT_CLIENT_SUB_TOPIC, QOS0, messageArrived);
require_noerr_string(rc, MQTT_reconnect, "ERROR: MQTT client subscribe err.");
mqtt_log("MQTT client subscribe success! recv_topic=[%s].", MQTT_CLIENT_SUB_TOPIC);
// 5. client loop for recv msg && keepalive
while (1){
no_mqtt_msg_exchange = true;
FD_ZERO(&readfds);
FD_SET(c.ipstack->my_socket, &readfds);
select(c.ipstack->my_socket + 1, &readfds, NULL, NULL, &t);
// recv msg from server
if (FD_ISSET( c.ipstack->my_socket, &readfds)){
rc = MQTTYield(&c, (int)MQTT_YIELD_TMIE);
require_noerr(rc, MQTT_reconnect);
no_mqtt_msg_exchange = false;
}
//if no msg exchange, we need to check ping msg to keep alive
if (no_mqtt_msg_exchange){
rc = keepalive(&c);
require_noerr_string(rc, MQTT_reconnect, "ERROR: keep alive err");
}
}  
MQTT_reconnect:
mqtt_log("Disconnect MQTT client, and reconnect after 5s, reason: mqtt_rc = %d, err = %d", rc, err);
mqtt_client_release(&c, &n);
sleep(5);
goto MQTT_start;
exit:
mqtt_log("EXIT: MQTT client exit with err = %d.", err);
mqtt_client_release(&c, &n);
pthread_exit(NULL);
}
static OSStatus mqtt_msg_publish(Client *c, const char* topic, char qos, char retained,
const unsigned char* msg,
uint32_t msg_len){
OSStatus err = kUnknownErr;
int ret = 0;
MQTTMessage publishData = MQTTMessage_publishData_initializer;
require(topic && msg_len && msg, exit);
//upload data qos0
publishData.qos = qos;
publishData.retained = retained;
publishData.payload = (void*)msg;
publishData.payloadlen = msg_len;
ret = MQTTPublish(c, topic, &publishData);
if (MQTT_SUCCESS == ret){
err = kNoErr;
}else{
err = kUnknownErr;
}
exit:
return err;
}
void user_send_cb(void* data){
OSStatus err = kNoErr;
s_MQTT_Data_Packet_Info* p_send_msg = (s_MQTT_Data_Packet_Info*)data;
require_noerr_string((p_send_msg == NULL), exit, "没有内存可用");
err = mqtt_msg_publish(&c, p_send_msg->topic, p_send_msg->qos, p_send_msg->retained,
p_send_msg->data,
p_send_msg->datalen);
require_noerr_string(err, exit, "publish失败");
mqtt_log("MQTT publish data success! send_topic=[%s], msg=[%ld][%s].\r\n", p_send_msg->topic, p_send_msg->datalen, p_send_msg->data);
no_mqtt_msg_exchange = false; // 在当前情况下，多发一次或少发一次，无关紧要，无需用互斥锁
exit:
if (p_send_msg != NULL){
free(p_send_msg);  
p_send_msg = NULL;
}
}
int main(void){
// void *rval;
OSStatus err = kNoErr;
lhead = (struct head*)malloc(sizeof(struct head));
STAILQ_INIT(lhead);
pthread_t mqtt_client_handle, work_thread_Handle/*, timer_thread_Handle*/;
// 默认堆栈大小为8M, 嵌入式里太大，重新设置
pthread_attr_t attr;
err = pthread_attr_init(&attr);
require_noerr_string(err, exit, "ERROR: Unable to init thread attr.");
err = pthread_attr_setstacksize(&attr, 16384);// 堆栈大小不能小于16384Byte 
require_noerr_string(err, exit, "ERROR: Unable to set thread size.");
err = pthread_create(&mqtt_client_handle, &attr, mqtt_client_thread, NULL);
require_noerr_string(err, exit, "ERROR: Unable to start the mqtt client thread.");
err = pthread_create(&work_thread_Handle, &attr, work_thread, NULL);
require_noerr_string(err, exit, "ERROR: Unable to start the work thread.");
// err = pthread_create(&timer_thread_Handle, NULL, timer_thread, NULL);
// require_noerr_string(err, exit, "ERROR: Unable to start the timer thread.");
// loop = uv_default_loop();
// uv_timer_t period_timer;
// uv_timer_init(loop, &period_timer);
// uv_timer_start(&period_timer, user_send_handler, 0, 2000);
// err = uv_run(loop, UV_RUN_DEFAULT); 
// require_noerr_string(err, exit, "ERROR: Unable to run uv loop.");
// struct timerval interval;
// struct itimerval timer;
pthread_join(work_thread_Handle, NULL);
pthread_join(mqtt_client_handle, NULL);   重点，当主线程没有其他可执行的循环时，一定要加此句
pthread_attr_destroy(&attr);
exit :
if (err != kNoErr){
app_log("ERROR: app thread exit err: %d", err);   
}
free(lhead);
lhead  = NULL;
return err;
}

Jetbrains全家桶1年46，售后保障稳定