1841
来源 | 嵌入式应用研究院
整理&排版 | 嵌入式应用研究院
在物联网相关的应用开发中或多或少都会用到MQTT，以下这个开源项目是我基于杰杰大佬的mqttclient项目进行二次封装的接口：
https://github.com/Yangyuanxin/EasyMqttClient
杰杰大佬的mqttclient项目：
https://github.com/jiejieTop/mqttclient
在封装之前，通过内存泄露工具定位排查得知调用mqtt_release和SALOF_LOG存在一些问题，好在mqtt_release的场景其实基本上不会用到，但还是有必要拿出来说说：
- Bug1(
platform_thread内存泄露)
mqttclient/platform/linux/platform_thread.c文件中的platform_thread_destroy函数并没有对线程封装函数中的thread所申请的内存进行释放，这样的后果是会在mqttclient调用mqtt_release函数的时候造成内存泄露。
- Bug2(实际待解决-可通过屏蔽宏定义解决)
mqttclient/common/log模块存在内存泄露，可通过关闭mqtt_config.h文件中的MQTT_LOG_IS_SALOF解决。
- Bug3(待解决-可暂时忽略)
在执行mqtt_release时，有机率出现core dump，但是mqtt_release场景在一般产品开发中并不常见，除非有特殊的需要。具体详见我提出的Issues:
https://github.com/jiejieTop/mqttclient/issues/60
除此之外，mqttclient用起来还是很爽的，不少开源项目和实际产品上都用了这套接口，非常稳定。
以下是我封装以后的接口就只有7个API，十分简单！分别是：
//MQTT初始化
EasyMqttClient_t *EasyMqttInit(EasyMqttAttr_t *Attr);
//MQTT反初始化
int EasyMqttUnInit(EasyMqttClient_t *Client);
//MQTT连接
int EasyMqttConnect(EasyMqttClient_t *Client);
//MQTT断开连接
int EasyMqttDisConnect(EasyMqttClient_t *Client);
//MQTT Topic订阅
int EasyMqttSubscribe(EasyMqttClient_t *Client, const char *Topic, enum EasyMqttQos_t Qos,
void (*Cb)(const char *Topic,char* Data,unsigned short Len));
//MQTT 解除订阅
int EasyMqttUnsubscribe(EasyMqttClient_t *Client, const char *Topic);
//MQTT Topic发布
int EasyMqttPublish(EasyMqttClient_t *Client, const char *Topic, enum EasyMqttQos_t Qos, char *Data, unsigned short Len);
其中EasyMqttInit函数将以下这些琐碎的过程，例如设置URL、设置端口号等过程用结构体EasyMqttAttr封装到了一起：
typedef struct EasyMqttAttr
{
char *Url;
char *Port;
char *ClientId;
char *Username;
char *Password;
}EasyMqttAttr_t;
//......................
mqtt_set_host;
mqtt_set_port;
mqtt_set_client_id;
mqtt_set_user_name;
mqtt_set_password;
mqtt_set_clean_session;
//调用的时候很简单
//1.
//2.定义一个结构体变量
example:
EasyMqttClient_t *Client = NULL;
EasyMqttAttr_t Attr =
{
.Url = "192.168.4.248",
.Port = "30157",
.ClientId = "EasyMqttMqtt",
.Username = "EasyMqtt",
.Password = "123456"
};
//3.调用EasyMqttInit函数
Client = EasyMqttInit(Client, &Attr);
//to do
//实现你的MQTT连接、订阅、分布等逻辑
//to do end
另外，它还实现了对不同订阅Topic的回调函数进行分开处理，让开发的逻辑更加清晰，也易于调试和解决问题，这个实现的机制是基于一个结构体数组来实现的，如下所示：
struct TopicHandler_t
{
//Topic
const char *Topic;
//Topic对应的回调函数
void (*CallBack)(const char *Topic,char* Data,unsigned short Len);
};
//结构体数组表，最大支持处理Topic的个数为MAX_TOPIC，该值默认为64
struct TopicHandler_t Table[MAX_TOPIC];
当调用EasyMqttSubscribe Topic订阅函数订阅一个Topic时，就会将这个Topic和它的回调添加到这个表里。当mqttclient接收到不同的Topic时，则会查表调用不同Topic所对应的回调函数，具体逻辑如下所示：
//Topic回调触发
static void TopicHandlerCallBack(void* client, message_data_t* Msg)
{
(void)client;
int Index = 0;
char *Topic = Msg->topic_name;
unsigned short Len = Msg->message->payloadlen;
char *Data = (char *)Msg->message->payload;
//上锁
pthread_mutex_lock(&Mutex);
//当接收到不同的Topic时，根据Topic找到对应的回调函数并进行调用
for(Index = 0; Index < sizeof(Table)/sizeof(Table[0]); Index++)
{
if(0 == strcmp(Msg->topic_name,Table[Index].Topic))
{
Table[Index].CallBack(Topic,Data,Len);
break;
}
}
//解锁
pthread_mutex_unlock(&Mutex);
}
//EasyMqttSubscribe Topic订阅函数
int EasyMqttSubscribe(EasyMqttClient_t *Client, const char *Topic, enum EasyMqttQos_t Qos,
void (*Cb)(const char *Topic, char* Data, unsigned short Len))
{
if(Index > MAX_TOPIC-1)
{
printf("Exceeds the maximum number of topics set:%d!\n", Index);
return -1;
}
Table[Index].Topic = Topic;
Table[Index].CallBack = Cb;
Index++;
return mqtt_subscribe(Client, Topic, (mqtt_qos_t)Qos, TopicHandlerCallBack);
}
具体使用方法可参考EasyMqtt.c中的EasyMqttTest函数。目前该项目仅在Linux项目上测试通过，后续将在不同的RTOS环境下进行测试。欢迎持续关注，也欢迎提Pr，共同让嵌入式MQTT应用开发变得更简单。
在Linux环境下使用本项目：
- 1、克隆本项目
git clone https://github.com/Yangyuanxin/EasyMqttClient.git
- 2、修改交叉编译工具链(默认为gcc)
如果你希望在嵌入式平台运行，则需要修改Makefile里的：
CROSS_COMPILE =
否则默认以gcc环境编译。
- 3、编译
make
- 4、执行
./a.out
其它环境：待测试。
