C语言网络编程（5）— MQTT通信_mqtt c语言

C语言网络编程（5）— MQTT通信

一、下载mqtt C语言库

首先我们要下载支持MQTT的C语言库，直接使用git命令拉取：

git clone https://github.com/eclipse/paho.mqtt.c.git
1

API文档介绍：http://www.eclipse.org/paho/files/mqttdoc/MQTTClient/html/index.html
安装一些需要使用的库：

sudo apt-get install libssl-dev build-essential libidn11-dev libidn11
1

然后我们进入下载好的源码目录，使用make命令编译得到我们需要的库文件

然后我们可以在paho.mqtt.c/build/output下可以找到如下的输出文件：

这里面的各个动态库的作用：

paho-mqtt3a  : 一般实际开发中就是使用这个，a表示的是异步消息推送（asynchronous）。
paho-mqtt3as : as表示的是 异步+加密（asynchronous+OpenSSL）。
paho-mqtt3c  : c 表示的应该是同步（Synchronize），一般性能较差，是发送+等待模式。
paho-mqtt3cs : cs表示的是同步+加密（asynchronous+OpenSSL）。
1
2
3
4

然后我们使用make install则是将生成的库文件移动到系统路径之下。

在paho.mqtt.c/src/samples目录下有一些示例：

二、编写程序连接到MQTT服务器

首先包含需要用到的头文件：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"
1
2
3
4

然后宏定义一些连接mqtt服务器需要的参数：

#define MQTT_Uri    "tcp://39.96.35.207:1883"   // MQTT服务器的地址和端口号
#define ClientId    "ubuntu16"                  // ClientId需要唯一
#define username    "ubuntu16"                  // 用户名
#define password    "123456"                    // 用户名对应的密码
1
2
3
4

然后我们可以开始编写main()函数，
首先，定义一个MQTT客户端结构体指针，用来接收后面创建的MQTT客户端结构体

    // 1、定义一个MQTT客户端结构体指针
    MQTTClient client;
1
2

然后使用MQTTClient_create()函数创建一个MQTT客户端，创建成功后客户端结构体将赋值给client；

    // 2、创建一个MQTT客户端
    if ((rc = MQTTClient_create(&client, MQTT_Uri, ClientId, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS)
    {
         printf("Failed to create mqtt client, return code %d\n", rc);
         exit(EXIT_FAILURE);
    }
1
2
3
4
5
6

接下来准备连接用的结构体

    // 3、创建一个MQTT连接配置结构体，并配置其参数
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.username = UserName;          // 用户名 
    conn_opts.password = PassWord;          // 用户名对应的密码
    conn_opts.keepAliveInterval = 60;       // 心跳时间
    conn_opts.cleansession = 1;             // 清除会话
1
2
3
4
5
6

然后设置回调函数，

	// 4、设置MQTT连接时的回调函数
    MQTTClient_setCallbacks(client, NULL, connect_lost, message_arrived, delivery_complete);
1
2

失去连接的回调函数

// 失去连接回调函数
void connect_lost(void *context, char *cause)
{
    printf("Connection lost,The reason: %s \n",cause);
}
1
2
3
4
5

配置好之后就可以连接到MQTT服务器了

    // 5、开始连接到MQTT服务器
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
1
2
3
4
5
6

我们先让其延时10秒以便于我们观察

    sleep(10);
1

因为我们要用到动态库文件paho-mqtt3c.so，使用在编译的时候要加上参数-lpaho-mqtt3c，我们编译，运行

然后打开EMQ控制台可以看到，已经连接成功了

三、发布主题

连接成功后就可以发布主题了，首先定义一个用来存储主题消息的结构体，然后将配置成我们需要发送的消息

    // 6、定义一个主题消息存储结构体
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    char mag_data[] = "I am Ubuntu16.";
    pubmsg.payload = mag_data;
    pubmsg.payloadlen = (int)strlen(mag_data);
    pubmsg.qos = 1;                 // qos等级为1 
    pubmsg.retained = 0;            // 服务器不保留消息
    MQTTClient_deliveryToken token; // 标记MQTT消息的值，用来检查消息是否发送成功

    // 7、发布主题信息
    if ((rc = MQTTClient_publishMessage(client, "ubuntu16_publish", &pubmsg, &token)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to publish message, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

然后实现主题发布成功的回调函数

// 主题发布成功回调函数
void delivery_complete(void *context, MQTTClient_deliveryToken dt)
{
    printf("publish topic success,token = %d \n", dt);
}
1
2
3
4
5

编译，运行代码，可以看到，发布消息成功：

使用MQTTfx来订阅ubuntu16_publish主题，可以收到消息：

四、订阅主题

接下来，我们订阅一个主题：

    // 8、订阅主题
    if ((rc = MQTTClient_subscribe(client, "ubuntu16_subscribe", 1)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to subscribe, return code %d\n", rc);
        rc = EXIT_FAILURE;
    }
1
2
3
4
5
6

然后实现收到主题信息回调函数

// 收到主题信息回调函数
int message_arrived(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    printf("Receive topic: %s,message data: \n", topicName);
    printf("%.*s\n", message->payloadlen, (char*)message->payload);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}
1
2
3
4
5
6
7
8
9

编译，运行，通过MQTTfx发布一个ubuntu16_subscribe主题消息，可以看到，接收到消息

四、代码

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

#define MQTT_Uri    "tcp://39.96.35.207:1883"   // MQTT服务器的地址和端口号
#define ClientId    "ubuntu16"                  // ClientId需要唯一
#define UserName    "ubuntu16"                  // 用户名
#define PassWord    "123456"                    // 用户名对应的密码

// 失去连接回调函数
void connect_lost(void *context, char *cause)
{
    printf("Connection lost,The reason: %s \n",cause);
}

// 收到主题信息回调函数
int message_arrived(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    printf("Receive topic: %s, message data: \n", topicName);
    printf("%.*s\n", message->payloadlen, (char*)message->payload);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

// 主题发布成功回调函数
void delivery_complete(void *context, MQTTClient_deliveryToken dt)
{
    printf("publish topic success,token = %d \n", dt);
}

int main(int argc, char* argv[])
{
    // 1、定义一个MQTT客户端结构体指针
    MQTTClient client;

    // 2、创建一个MQTT客户端
    int rc;
    if ((rc = MQTTClient_create(&client, MQTT_Uri, ClientId, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to create client, return code %d\n", rc);
        exit(EXIT_FAILURE);
        goto exit;
    }

    // 3、创建一个MQTT连接配置结构体，并配置其参数
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.username = UserName;          // 用户名 
    conn_opts.password = PassWord;          // 用户名对应的密码
    conn_opts.keepAliveInterval = 60;       // 心跳时间
    conn_opts.cleansession = 1;             // 清除会话
    
    // 4、设置MQTT连接时的回调函数
    if ((rc = MQTTClient_setCallbacks(client, NULL, connect_lost, message_arrived, delivery_complete)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to set callbacks, return code %d\n", rc);
        rc = EXIT_FAILURE;
        goto destroy_exit;
    }

    // 5、开始连接到MQTT服务器
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
        goto destroy_exit;
    }

    // 6、定义一个主题消息存储结构体
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    char mag_data[] = "I am Ubuntu16.";
    pubmsg.payload = mag_data;
    pubmsg.payloadlen = (int)strlen(mag_data);
    pubmsg.qos = 1;                 // qos等级为1 
    pubmsg.retained = 0;            // 服务器不保留消息
    MQTTClient_deliveryToken token; // 标记MQTT消息的值，用来检查消息是否发送成功

    // 7、发布主题信息
    if ((rc = MQTTClient_publishMessage(client, "ubuntu16_publish", &pubmsg, &token)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to publish message, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    // 8、订阅主题
    if ((rc = MQTTClient_subscribe(client, "ubuntu16_subscribe", 1)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to subscribe, return code %d\n", rc);
        rc = EXIT_FAILURE;
    }

    // 9、等待输入'Q'或'q'退出
    printf("Press Q or q + <Enter> to quit\n\n");
    int ch;
    do
    {
            ch = getchar();
    } while (ch!='Q' && ch != 'q');

    if ((rc = MQTTClient_unsubscribe(client, "ubuntu16_subscribe")) != MQTTCLIENT_SUCCESS)
    {
            printf("Failed to unsubscribe, return code %d\n", rc);
            rc = EXIT_FAILURE;
    }

    // 10、断开连接
    if ((rc = MQTTClient_disconnect(client, 10000)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to disconnect, return code %d\n", rc);
        rc = EXIT_FAILURE;
    }

destroy_exit:
    MQTTClient_destroy(&client);
exit:
    return rc;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120