MQTT标准协议-MQTT C Client Library

MQTT标准协议

一、Linux安装MQTT C library

GitHub: https://github.com/eclipse/paho.mqtt.c

git clone https://github.com/eclipse/paho.mqtt.c.git
cd paho.mqtt.git
make
sudo make install

二、代码编写

Paho Asynchronous MQTT C Client Library手册链接地址：

Asynchronous MQTT client library for C​www.eclipse.org

 

Paho MQTT C Client Library手册链接地址：

MQTT Client library for C​www.eclipse.org

 

以下是mqtt.c测试程序，依次按照client连接服务器、订阅主题和发布消息流程实现，订阅接收数据采用回调函数，具备异步通信功能。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <MQTTAsync.h>
#include <MQTTClient.h>

#define ADDRESS     "tcp://localhost:1883"
#define CLIENTID    "ExampleClientPub"
#define TOPIC       "mqtt"
#define PAYLOAD     "Hello World!"
#define QOS         1
#define TIMEOUT     10000L

static MQTTClient client = NULL;
volatile MQTTClient_deliveryToken deliveredtoken;

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char* payloadptr;

    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");

    payloadptr = message->payload;
    for(i=0; i<message->payloadlen; i++)
    {
        putchar(*payloadptr++);
    }
    putchar('\n');

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);

    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

int mqtt_client_publish(char *top, int qos, char *msg, int len)
{
    MQTTClient_deliveryToken token;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;

    if (client == NULL)
        return -1;

    pubmsg.payload = msg;
    pubmsg.payloadlen = len;
    pubmsg.qos = qos;
    pubmsg.retained = 0;
    deliveredtoken = 0;
    MQTTClient_publishMessage(client, top, &pubmsg, &token);
    printf("Waiting for publication of %s\n"
            "on topic %s for client\n",
            msg, top);
    while (deliveredtoken != token);
    return 0;
}

int mqtt_client_subscribe(char *top, int qos)
{
    if (client == NULL)
	return -1;

    MQTTClient_subscribe(client, top, qos);
    return 0;
}

int mqtt_client_open()
{
    int rc;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
   
    MQTTClient_create(&client, ADDRESS, CLIENTID,        MQTTCLIENT_PERSISTENCE_NONE, NULL);
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        return -1;
    }
    printf("Success to connect!\n");
    return 0;
}

int main(int argc, char* argv[])
{
    char *str = "hello world!";

    mqtt_client_open();
    mqtt_client_subscribe(TOPIC, QOS);
    mqtt_client_publish(TOPIC, QOS, str, strlen(str));
    sleep(1);

    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    
    return 0;
}

三、编译程序

gcc -o mqtt mqtt.c -lpaho-mqtt3c -lpaho-mqtt3a

arm-linux-gnueabihf-gcc -o mqtt mqtt_test.c -lpaho-mqtt3c -lpaho-mqtt3a -L ./lib -I./inc

必须要加入动态链接库选项，否则编译可能会出现以下错误：

/tmp/ccIiNRZp.o: In function `msgarrvd':
mqtt.c:(.text+0xc5): undefined reference to `MQTTClient_freeMessage'
mqtt.c:(.text+0xd1): undefined reference to `MQTTClient_free'
/tmp/ccIiNRZp.o: In function `mqtt_client_publish':
mqtt.c:(.text+0x1e2): undefined reference to `MQTTClient_publishMessage'
/tmp/ccIiNRZp.o: In function `mqtt_client_subscribe':
mqtt.c:(.text+0x260): undefined reference to `MQTTClient_subscribe'
/tmp/ccIiNRZp.o: In function `mqtt_client_open':
mqtt.c:(.text+0x358): undefined reference to `MQTTClient_create'
mqtt.c:(.text+0x37c): undefined reference to `MQTTClient_setCallbacks'
mqtt.c:(.text+0x3a9): undefined reference to `MQTTClient_connect'
/tmp/ccIiNRZp.o: In function `main':
mqtt.c:(.text+0x46f): undefined reference to `MQTTClient_disconnect'
mqtt.c:(.text+0x479): undefined reference to `MQTTClient_destroy'
collect2: error: ld returned 1 exit status

四、数据结构含义

MQTTClient_connectOptions定义

typedef struct
{
char struct_id[4];    //结构体的识别序列，必须为MQTC
int struct_version;   //结构体版本
/**
在0,1,2,3,4,5中取值：
0-表示没有SSL选项且没有serverURIs；
1-表示没有serverURIs；
2-表示没有MQTTVersion
3-表示没有返回值；
4-表示没有二进制密码选项
*/
int keepAliveInterval;
/**
在这段时间内没有数据相关的消息时，客户端发送一个非常小的MQTT“ping”消息，服务器将会确认这个消息
*/
int cleansession;
/**
当cleansession为true时，会话状态信息在连接和断开连接时被丢弃。 将cleansession设置为false将保留会话状态信息
*/
int reliable;
/*
将该值设置为true意味着必须完成发布的消息（已收到确认），才能发送另一个消息
*/
MQTTClient_willOptions* will;
/*
如果程序不使用最后的意愿和遗嘱功能，请将此指针设置为NULL。
*/
const char* username;//用户名
const char* password;//密码
int connectTimeout;//允许尝试连接的过时时间
int retryInterval;//尝试重连的时间
MQTTClient_SSLOptions* ssl;
/*
如果程序不使用最后的ssl,请将此指针设置为NULL。
*/
int serverURIcount;

char* const* serverURIs;
/*
连接服务器的url，以protocol：// host：port为格式
*/
int MQTTVersion;
/*
MQTT的版本,MQTTVERSION_3_1（3），MQTTVERSION_3_1_1 (4) 
*/
struct
{
const char* serverURI;   
int MQTTVersion;     
int sessionPresent;  
} returned;
  struct {
  int len;            
const void* data;  
} binarypwd;
} MQTTClient_connectOptions;

MQTTClient_message

typedef struct
{
    char struct_id[4];//结构体的识别序列，必须为MQTM
    int struct_version;//结构体的版本，必须为0
    int payloadlen;//MQTT信息的长度
    void* payload;//指向消息负载的指针
    int qos;//服务质量
    int retained;//保留标志
    int dup;dup//标志指示这个消息是否是重复的。 只有在收到QoS1消息时才有意义。 如果为true，则客户端应用程序应采取适当的措施来处理重复的消息。
    int msgid;//消息标识符通常保留供MQTT客户端和服务器内部使用。
} MQTTClient_message;

五、操作函数

DLLExport int MQTTClient_create(

MQTTClient * handle,

const char * serverURI,

const char * clientId,

int persistence_type,

void * persistence_context

)

作用：
该函数创建了一个用于连接到特定服务器，使用特定持久存储的MQTT客户端。

参数：
handle 指向MQTT客户端句柄的指针。句柄被成功从函数中返回的客户端引用所填充
serverURI 以空结尾的字符串，其指定客户端将连接到的服务器。其格式为protocol://host:port。现在的（protocol）协议必须是tcp或ssl，而host可以指定为IP地址或域名。例如, 要使用默认 MQTT 端口连接到本地计算机上运行的服务器, 请指定为 tcp://localhost:1883。
clientId 客户端标识符（clientId）是一个以空结尾的 UTF-8 编码字符串，客户端连接到服务器时将它传递过去。
persistence_type 客户端所使用的持久类型。MQTTCLIENT_PERSISTENCE_NONE-使用内存持久化。如果客户端运行的设备或系统出故障或关闭, 则任何正在运行的消息的当前状态都将丢失, 甚至在 QoS1 和 QoS2 中也可能无法传递某些消息; MQTTCLIENT_PERSISTENCE_DEFAULT-使用默认的持久化机制（文件系统）。正在运行消息的状态被保存在持久存储中，以便在意外出现时对消息的丢失提供一些保护; MQTTCLIENT_PERSISTENCE_USER-使用程序指定的持久化实现。使用这种类型，应用程序可对持久化机制进行控制，应用程序必须实现MQTTClient_persistence 接口。
persistence_context 如果应用程序使用的是MQTTCLIENT_PERSISTENCE_NONE持久化，该参数不使用，而且值应该设置为NULL。对于MQTTCLIENT_PERSISTENCE_DEFAULT持久化，应该设置持久化目录的位置（如果设置为NULL，则使用工作目录作为持久化目录）。使用MQTTCLIENT_PERSISTENCE_USER持久化，则将此参数指向有效的MQTTClient_persistence结构。

 

DLLExport int MQTTClient_setCallbacks (

MQTTClient handle,

void * context,

MQTTClient_connectionLost * cl,

MQTTClient_messageArrived * ma,

MQTTClient_deliveryComplete * dc

)

作用：
该函数为特定的客户端创建回调函数。如果您的客户端应用程序不使用特定的回调函数，请将相关参数设置为NULL。 调用MQTTClient_setCallbacks（）使客户端进入多线程模式。 任何必要的消息确认和状态通信都在后台处理，而不需要客户端应用程序的任何干预。
注意：在调用该函数时，MQTT客户端必须断开连接。（即先要调用该函数在连接客户端）。

参数：
handle 指向MQTT客户端句柄的指针。句柄被成功从函数中返回的客户端引用所填充
context 指向任何应用程序特定上下文的指针。 上下文指针被传递给每个回调函数，以提供对回调中的上下文信息的访问。
cl 指向MQTTClient_connectionLost()回调函数的指针。 如果您的应用程序不处理断开连接，您可以将其设置为NULL。
ma 指向MQTTClient_messageArrived()回调函数的指针。 当您调用MQTTClient_setCallbacks()时，必须指定此回调函数。
dc 指向MQTTClient_deliveryComplete()回调函数的指针。 如果您的应用程序同步发布，或者您不想检查是否成功发送，则可以将其设置为NULL。

 

DLLExport int MQTTClient_connect (

MQTTClient handle,

MQTTClient_connectOptions * options

)

作用：
此函数尝试使用指定的选项将先前创建的客户端连接到MQTT服务器。

参数：
handle 指向MQTT客户端句柄的指针。句柄被成功从函数中返回的客户端引用所填充
options 指向有效的MQTTClient_connectOptions结构的指针。

返回值：
0 连接成功
1 拒绝连接：不可接受的协议版本。
2 拒绝连接：标识符被拒绝。
3 拒绝连接：服务器不可用。
4 拒绝连接：用户名或密码错误。
5 拒绝连接：未经授权。
6 保留给未来用。

 

DLLExport int MQTTClient_subscribe (

MQTTClient handle,

const char * topic,

int qos

)

作用：
此功能尝试将客户订阅到单个主题，该主题可能包含通配符。 此函数还指定服务质量。

参数：
handle 指向MQTT客户端句柄的指针。句柄被成功从函数中返回的客户端引用所填充
topic 订阅的主题，可使用通配符。
qos 订阅的请求服务质量

 

DLLExport int MQTTClient_publishMessage (

MQTTClient handle,

const char * topicName,

MQTTClient_message * msg,

MQTTClient_deliveryToken * dt

)

作用：
此功能尝试将客户订阅到单个主题，该主题可能包含通配符。 此函数还指定服务质量。

参数：
handle 指向MQTT客户端句柄的指针。句柄被成功从函数中返回的客户端引用所填充
topicName 与信息相关的主题。
msg 指向有效的 MQTTClient_message 结构的指针, 其中包含要发布消息的有效负载和属性
dt 指向MQTTClient_deliveryToken的指针。当函数成功返回时，dt会被赋值为代表消息的token。如果程序中没有使用传递token，将其设置为NULL。

 

DLLExport int MQTTClient_waitForCompletion (

MQTTClient handle,

MQTTClient_deliveryToken dt,

unsigned long timeout

)

作用：
客户端应用程序调用此函数来将主线程的执行与消息的完成发布同步。 被调用时，MQTTClient_waitForCompletion()阻塞执行，直到消息成功传递或已超过指定的时间。

参数：
handle 指向MQTT客户端句柄的指针。句柄被成功从函数中返回的客户端引用所填充
dt 代表消息的MQTTClient_deliveryToken用来检测是否成功传递。传递token由发布函数MQTTClient_publish () 和 MQTTClient_publishMessage ()所产生。
timeout 等待的最大毫秒数。

返回值 ：
消息成功传递则返回MQTTCLIENT_SUCCESS(0) ，如果时间已过期或检测token时出问题，则返回错误码。

本文来自： https://zhuanlan.zhihu.com/p/54450815 

双龙子木

一位具有激情的奋斗者

另一篇博客也介绍类似的MQTT： https://blog.csdn.net/vincent_yuan89/article/details/84306001

协议：

http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html

源码：

https://os.mbed.com/teams/mqtt/code/MQTTPacket/

git 源码

https://github.com/eclipse/paho.mqtt.c