MQTT客户端应用编程及接口分析
MQTT协议简介
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的,这些特点使它适用范围非常广泛。
客户端服务端安装
1.安装
sudo apt-add-repository ppa:mosquitto-dev/mosquitto-ppasudo
sudo apt-get install mosquitto mosquitto-clients
- 测试
mosquitto_sub -h test.mosquitto.org -t "#" -v
如果发现大量的消息滚动刷屏,说明该客户端链接成功
客户端开发包
SDK包下载地址:
eclipse/paho.mqtt.c
编译及生产lib和头文件
cmake -DMAKE_INSTALL_PREFIX=./output
make && make install
API接口介绍
接口类型 | 头文件 | 备注 |
---|---|---|
Paho MQTT C Client Library | MQTTClient.h | 非现程安全 |
Paho Asynchronous MQTT C Client Library | MQTTAsync.h | 线程安全 |
数据结构
typedef void* MQTTClient;
typedef void* MQTTAsync;
MQTTClient 库接口
MQTTClient_create() : MQTTClient.h
int MQTTClient_create(MQTTClient *handle,
const char *serverURI,
const char *clientId,
int persistence_type,
void * persistence_context
)
此函数创建一个 MQTT 客户端,以便连接到指定的服务器并使用指定的持久性存储(参考 MQTTClient_persistence 函数)。另外,与创建函数对应的是销毁函数 MQTTClient_destroy()。
MQTTClient_connect() : MQTTClient.h
int MQTTClient_connect(MQTTClient handle,
MQTTClient_connectOptions *options
)
此函数尝试使用指定的选项将以前创建的客户端(调用 MQTTClient_create() 函数)连接到 MQTT 服务器。如果要启用异步消息和状态通知,调用该函数之前必须先调用 MQTTClient_setCallbacks() 函数。
MQTTClient_destroy() : MQTTClient.h
void MQTTClient_destroy(MQTTClient *handle)
此函数用于释放分配给 MQTT 客户端的内存(调用 MQTTClient_create() 函数)。当不再需要客户端时,应该调用它。
4.MQTTClient_disconnect() : MQTTClient.h
int MQTTClient_disconnect(MQTTClient handle, int timeout)
此函数尝试断开客户端与 MQTT 服务器的连接。为了让客户端有时间完成对调用此函数时正在传递的消息的处理,需要指定一个超时期限。当超时期限过期时,即使仍有未完成的消息确认,客户端也会断开连接。下一次客户端连接到同一服务器时,任何未完成的 QoS 1 或 2 消息将根据前一个连接和新连接的清理设置重试(参考 MQTTClient_connectOptions.cleansession 和 MQTTClient_connect())。
MQTTClient_free() : MQTTClient.h
void MQTTClient_free(void *ptr)
此函数释放 MQTT C Client 库分配的内存,特别是主题名称。当 client 库和应用程序使用不同版本的 C 编译器编译时,在 Windows 上需要这样做。因此,在释放任何 MQTT C Client 分配的内存时,始终使用此函数是一个很好的策略。
MQTTClient_freeMessage() : MQTTClient.h
void MQTTClient_freeMessage(MQTTClient_message ** msg)
此函数释放分配给 MQTT 消息的内存,包括分配给消息有效负载的附加内存。当完全处理消息时,客户端应用程序调用此函数。重要提示:此函数不释放分配给消息主题字符串的内存。客户端应用程序负责使用 MQTTClient_free() 库函数释放此内存。
MQTTClient_isConnected() : MQTTClient.h
int MQTTClient_isConnected(MQTTClient handle)
此函数允许客户端应用程序测试某个 client 与 MQTT 服务器连接情况,已连接返回 true,否则返回 false。
MQTTClient_publish() : MQTTClient.h
int MQTTClient_publish(MQTTClient handle,
const char * topicName,
int payloadlen,
const void * payload,
int qos,
int retained,
MQTTClient_deliveryToken *dt
)
此函数试图将消息发布到给定主题(另请参见 MQTTClient_publishMessage 函数)。当此函数成功返回时,将发出 MQTTClient_deliveryToken。如果客户端应用程序需要测试 QoS1 和 QoS2 消息的成功传递,可以异步或同步地完成(参考 异步与同步客户端应用程序、 MQTTClient_waitForCompletion 和 MQTTClient_deliveryComplete 函数)。
MQTTClient_publishMessage() : MQTTClient.h
int MQTTClient_publishMessage(MQTTClient handle,
const char * topicName,
MQTTClient_message * msg,
MQTTClient_deliveryToken *dt
)
此函数试图将消息发布到给定主题(另请参见 MQTTClient_publish 函数)。当此函数成功返回时,将发出一个 MQTTClient_deliveryToken 令牌。如果客户端应用程序需要测试 QoS1 和 QoS2 消息的成功传递,可以异步或同步地完成(参考 异步与同步客户端应用程序、 MQTTClient_waitForCompletion 和 MQTTClient_deliveryComplete 函数)。
MQTTClient_receive() : MQTTClient.h
int MQTTClient_receive(MQTTClient handle,
char ** topicName,
int * topicLen,
MQTTClient_message ** message,
unsigned long timeout
)
此函数执行传入消息的同步接收。只有当客户端应用程序没有设置回调方法以支持消息的异步接收时,才应该使用它(请参见 异步与同步客户端应用程序 以及 MQTTClient_setCallbacks 函数)。使用此函数可以编写单线程客户端订阅者应用程序。当调用时,此函数将阻塞,直到下一条消息到达或指定的超时过期为止(另请参见 MQTTClient_yield 函数)。
重要提示:应用程序在处理完成后必须释放分配给主题和消息的内存(请参见 MQTTClient_freeMessage 函数)。
MQTTClient_setCallbacks() : MQTTClient.h
int MQTTClient_setCallbacks(MQTTClient handle,
void * context,
MQTTClient_connectionLost * cl,
MQTTClient_messageArrived * ma,
MQTTClient_deliveryComplete * dc
)
此函数设置特定客户端的回调函数。如果客户端应用程序不使用特定的回调,请将相关参数设置为 NULL。调用 MQTTClient_setCallbacks() 将客户端置于多线程模式。任何必要的消息确认和状态通信都在后台处理,不需要客户端应用程序的任何干预。更多信息请参见 异步与同步客户端应用程序。
注意:在调用此函数时,MQTT 客户端必须断开连接。
MQTTClient_setDisconnected() : MQTTClient.h
int MQTTClient_setDisconnected(MQTTClient handle,
void * context,
MQTTClient_disconnected *co
)
设置客户端的 MQTTClient_disconnected() 回调函数。如果从服务器接收到断开连接的数据包,将调用这个函数。该函数只对 MQTT V5 及以上版本有效。
MQTTClient_setPublished() : MQTTClient.h
int MQTTClient_setPublished(MQTTClient handle,
void * context,
MQTTClient_published * co
)
MQTTAsync 库接口
MQTTAsync_connect() :
MQTTAsync.hMQTTAsync_create() :
MQTTAsync.hMQTTAsync_createWithOptions() :
MQTTAsync.hMQTTAsync_destroy() :
MQTTAsync.hMQTTAsync_disconnect() :
MQTTAsync.hMQTTAsync_free() :
MQTTAsync.hMQTTAsync_freeMessage() :
MQTTAsync.hMQTTAsync_getPendingTokens() :
MQTTAsync.hMQTTAsync_getVersionInfo() :
MQTTAsync.hMQTTAsync_global_init() :
MQTTAsync.hMQTTAsync_isComplete() :
MQTTAsync.hMQTTAsync_isConnected() :
MQTTAsync.hMQTTAsync_malloc() :
MQTTAsync.hMQTTAsync_reconnect() :
MQTTAsync.hMQTTAsync_send() :
MQTTAsync.hMQTTAsync_sendMessage() :
MQTTAsync.hMQTTAsync_setAfterPersistenceRead() :
MQTTAsync.hMQTTAsync_setBeforePersistenceWrite() :
MQTTAsync.hMQTTAsync_setCallbacks() :
MQTTAsync.hMQTTAsync_setConnected() :
MQTTAsync.hMQTTAsync_setConnectionLostCallback() :
MQTTAsync.hMQTTAsync_setDeliveryCompleteCallback() :
MQTTAsync.hMQTTAsync_setDisconnected() :
MQTTAsync.hMQTTAsync_setMessageArrivedCallback() :
MQTTAsync.hMQTTAsync_setTraceCallback() :
MQTTAsync.hMQTTAsync_setTraceLevel() :
MQTTAsync.hMQTTAsync_setUpdateConnectOptions() :
MQTTAsync.hMQTTAsync_strerror() :
MQTTAsync.hMQTTAsync_subscribe() :
MQTTAsync.hMQTTAsync_subscribeMany() :
MQTTAsync.hMQTTAsync_unsubscribe() :
MQTTAsync.hMQTTAsync_unsubscribeMany() :
MQTTAsync.hMQTTAsync_waitForCompletion() :
MQTTAsync.hMQTTProperties_add() :
MQTTProperties.hMQTTProperties_copy() :
MQTTProperties.hMQTTProperties_free() :
MQTTProperties.hMQTTProperties_getNumericValue() :
MQTTProperties.hMQTTProperties_getNumericValueAt() :
MQTTProperties.hMQTTProperties_getProperty() :
MQTTProperties.hMQTTProperties_getPropertyAt() :
MQTTProperties.hMQTTProperties_hasProperty() :
MQTTProperties.hMQTTProperties_len() :
MQTTProperties.hMQTTProperties_propertyCount() :
MQTTProperties.hMQTTProperties_read() :
MQTTProperties.hMQTTProperties_write() :
MQTTProperties.hMQTTProperty_getType() :
MQTTProperties.hMQTTPropertyName() :
MQTTProperties.hMQTTReasonCode_toString() :
MQTTReasonCodes.h
###代码使用案例
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTAsync.h"
#if !defined(_WIN32)
#include <unistd.h>
#else
#include <windows.h>
#endif
#if defined(_WRS_KERNEL)
#include <OsWrapper.h>
#endif
#define ADDRESS "tcp://mqtt.eclipseprojects.io:1883"
#define CLIENTID "ExampleClientPub"
#define TOPIC "MQTT Examples"
#define PAYLOAD "Hello World!"
#define QOS 1
#define TIMEOUT 10000L
int finished = 0;
void connlost(void *context, char *cause)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
int rc;
printf("\nConnection lost\n");
printf(" cause: %s\n", cause);
printf("Reconnecting\n");
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start connect, return code %d\n", rc);
finished = 1;
}
}
void onDisconnectFailure(void* context, MQTTAsync_failureData* response)
{
printf("Disconnect failed\n");
finished = 1;
}
void onDisconnect(void* context, MQTTAsync_successData* response)
{
printf("Successful disconnection\n");
finished = 1;
}
void onSendFailure(void* context, MQTTAsync_failureData* response)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
int rc;
printf("Message send failed token %d error code %d\n", response->token, response->code);
opts.onSuccess = onDisconnect;
opts.onFailure = onDisconnectFailure;
opts.context = client;
if ((rc = MQTTAsync_disconnect(client, &opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start disconnect, return code %d\n", rc);
exit(EXIT_FAILURE);
}
}
void onSend(void* context, MQTTAsync_successData* response)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
int rc;
printf("Message with token value %d delivery confirmed\n", response->token);
opts.onSuccess = onDisconnect;
opts.onFailure = onDisconnectFailure;
opts.context = client;
if ((rc = MQTTAsync_disconnect(client, &opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start disconnect, return code %d\n", rc);
exit(EXIT_FAILURE);
}
}
void onConnectFailure(void* context, MQTTAsync_failureData* response)
{
printf("Connect failed, rc %d\n", response ? response->code : 0);
finished = 1;
}
void onConnect(void* context, MQTTAsync_successData* response)
{
MQTTAsync client = (MQTTAsync)context;
MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
int rc;
printf("Successful connection\n");
opts.onSuccess = onSend;
opts.onFailure = onSendFailure;
opts.context = client;
pubmsg.payload = PAYLOAD;
pubmsg.payloadlen = (int)strlen(PAYLOAD);
pubmsg.qos = QOS;
pubmsg.retained = 0;
if ((rc = MQTTAsync_sendMessage(client, TOPIC, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start sendMessage, return code %d\n", rc);
exit(EXIT_FAILURE);
}
}
int messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* m)
{
/* not expecting any messages */
return 1;
}
int main(int argc, char* argv[])
{
MQTTAsync client;
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
int rc;
if ((rc = MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTASYNC_SUCCESS)
{
printf("Failed to create client object, return code %d\n", rc);
exit(EXIT_FAILURE);
}
if ((rc = MQTTAsync_setCallbacks(client, NULL, connlost, messageArrived, NULL)) != MQTTASYNC_SUCCESS)
{
printf("Failed to set callback, return code %d\n", rc);
exit(EXIT_FAILURE);
}
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
conn_opts.onSuccess = onConnect;
conn_opts.onFailure = onConnectFailure;
conn_opts.context = client;
if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
{
printf("Failed to start connect, return code %d\n", rc);
exit(EXIT_FAILURE);
}
printf("Waiting for publication of %s\n"
"on topic %s for client with ClientID: %s\n",
PAYLOAD, TOPIC, CLIENTID);
while (!finished)
#if defined(_WIN32)
Sleep(100);
#else
usleep(10000L);
#endif
MQTTAsync_destroy(&client);
return rc;
}