利用mosquitto库实现客户端上报云平台首先需要连接阿里云测试接下来就用mosquitto库自己实现mqtt客户端

104 阅读 0 评论 69 点赞

我是靠谱客的博主忧郁飞机，这篇文章主要介绍利用mosquitto库实现客户端上报云平台首先需要连接阿里云测试接下来就用mosquitto库自己实现mqtt客户端，现在分享给大家，希望可以做个参考。

这里做的是上报阿里云平台

首先需要连接阿里云测试

这里我也是参考别人博客一步步做的...所以还是直接贴出链接：

https://blog.csdn.net/qq997758497/article/details/90757307

按照上面博客成功用MQTT.fx订阅与发布之后，接下来看一下我们得到的东西

订阅和发布的主题，这里用/sys/a1R4lbARNLO/${deviceName}/thing/event/property/post发布主题还有/sys/a1R4lbARNLO/${deviceName}/thing/service/property/set订阅主题这两个来测试，当然也可以用其他的主题

比起刚刚的产品主题Topic类，功能更像是主题

使用在线调试，发布消息MQTT.fx接收到的消息以Json格式打印出来

阿里云的上报会比较严格，还需要用相同的Json格式发布阿里云平台才能收得到，将订阅收到的Json格式消息复制到发布端去发布。

最少也需要以如下格式：

复制代码

1
2
3
4
5
6
7
8
9
{

  "params" : {

    "Data" : "Test"

  },

}

阿里云平台才接收数据显示对应功能的数据

有了这些东西之后就可以使用mosquitto连接阿里云平台了，用mosquitto_sub订阅。

用mosquitto_pub发布的时候载荷不符合Json格式，上报的数据对应的功能处并没有收到，可能是我还没完全掌握mosquitto_pub的用法吧

接下来就用mosquitto库自己实现mqtt客户端

使用了开源的iniparser库来操作配置文件，对于这些需要很多参数的程序，用一个配置文件来解决就很方便，可以了解一下iniparser库：

https://wenku.baidu.com/view/fba7c147c4da50e2524de518964bcf84b8d52d1a.html

一开始需要修改配置文件，配置文件默认上报到我的阿里云平台

订阅需要修改host，阿里云平台设备的username，passwd，客户端ID client_id，修改[sub_topic]topic来订阅主题，阿里云是通过json格式订阅和发布不同功能的mqtt报文，修改[ali_json]identifier标识符id等。

获取连接配置:

复制代码

int gain_mqtt_conf(char *ini_path,st_mqtt *mqtt, int type)

{

dictionary          *ini = NULL;

const char          *hostname ;

int                 port ;

const char          *username ;

const char          *passwd ;

const char          *clientid ;

const char          *topic ;

const char          *method ;

const char          *json_id ;

const char          *identifier ;

const char          *version ;

if(!ini_path || !mqtt)

{

printf("invail input parameter in %sn", __FUNCTION__) ;

return -1 ;

}

ini = iniparser_load(ini_path); //Load ini file get mqtt conf

if ( ini == NULL ) {

printf("iniparser_load error!n");

return -2;

}

hostname = iniparser_getstring(ini, "mqtt_server_addr:host", DEFAULT_HOSTNAME);

port = iniparser_getint(ini, "mqtt_server_addr:port", DEFAULT_PORT);

username = iniparser_getstring(ini, "user_passwd:username", DEFAULT_USERNAME);

passwd = iniparser_getstring(ini, "user_passwd:passwd", DEFAULT_PASSWD);

clientid = iniparser_getstring(ini, "client_id:id", DEFAULT_CLIENTID);

identifier = iniparser_getstring(ini,"json:identifier", DEFAULT_IDENTIFIER) ;

if(type == TYPE_SUB)

topic = iniparser_getstring(ini, "sub_topic:topic", DEFAULT_SUBTOPIC) ;

else if(type == TYPE_PUB)

{

topic = iniparser_getstring(ini,"pub_topic:topic", DEFAULT_PUBTOPIC) ;

method = iniparser_getstring(ini,"json:method", DEFAULT_METHOD) ;

json_id = iniparser_getstring(ini,"json:id", DEFAULT_JSONID) ;

version = iniparser_getstring(ini,"json:version", DEFAULT_VERSION) ;

}

strncpy(mqtt->hostname, hostname,BUF_SIZE) ;

mqtt->port = port ;

strncpy(mqtt->username, username,BUF_SIZE) ;

strncpy(mqtt->passwd, passwd,BUF_SIZE) ;

strncpy(mqtt->clientid, clientid,BUF_SIZE) ;

strncpy(mqtt->topic, topic,BUF_SIZE) ;

strncpy(mqtt->identifier, identifier,BUF_SIZE) ;

if(type == TYPE_PUB)

{

strncpy(mqtt->method, method,BUF_SIZE) ;

strncpy(mqtt->json_id, json_id,BUF_SIZE) ;

strncpy(mqtt->version, version,BUF_SIZE) ;

}

#if 0

printf(" MQTT hostname[%s]:%d,n clientid:%s,n username:%s, passwd:%s,n topic:%sn",

mqtt->hostname, mqtt->port, mqtt->clientid, mqtt->username, mqtt->passwd, mqtt->topic);

#endif

iniparser_freedict(ini); //Relaese ini file

return 0 ;

}

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
121
122
123
int gain_mqtt_conf(char *ini_path,st_mqtt *mqtt, int type)

{

    dictionary          *ini = NULL;

    const char          *hostname ;

    int                 port ;

    const char          *username ;

    const char          *passwd ;

    const char          *clientid ;

    const char          *topic ;

    const char          *method ;

    const char          *json_id ;

    const char          *identifier ;

    const char          *version ;



    if(!ini_path || !mqtt)

    {

        printf("invail input parameter in %sn", __FUNCTION__) ;

        return -1 ;

    }

    ini = iniparser_load(ini_path); //Load ini file get mqtt conf

    if ( ini == NULL ) {

        printf("iniparser_load error!n");

        return -2;

    }

    hostname = iniparser_getstring(ini, "mqtt_server_addr:host", DEFAULT_HOSTNAME);

    port = iniparser_getint(ini, "mqtt_server_addr:port", DEFAULT_PORT);

    username = iniparser_getstring(ini, "user_passwd:username", DEFAULT_USERNAME);

    passwd = iniparser_getstring(ini, "user_passwd:passwd", DEFAULT_PASSWD);

    clientid = iniparser_getstring(ini, "client_id:id", DEFAULT_CLIENTID);

    identifier = iniparser_getstring(ini,"json:identifier", DEFAULT_IDENTIFIER) ;

    if(type == TYPE_SUB)

        topic = iniparser_getstring(ini, "sub_topic:topic", DEFAULT_SUBTOPIC) ;

    else if(type == TYPE_PUB)

    {

        topic = iniparser_getstring(ini,"pub_topic:topic", DEFAULT_PUBTOPIC) ;

        method = iniparser_getstring(ini,"json:method", DEFAULT_METHOD) ;

        json_id = iniparser_getstring(ini,"json:id", DEFAULT_JSONID) ;

        version = iniparser_getstring(ini,"json:version", DEFAULT_VERSION) ;



    }

    strncpy(mqtt->hostname, hostname,BUF_SIZE) ;

    mqtt->port = port ;



    strncpy(mqtt->username, username,BUF_SIZE) ;

    strncpy(mqtt->passwd, passwd,BUF_SIZE) ;

    strncpy(mqtt->clientid, clientid,BUF_SIZE) ;

strncpy(mqtt->topic, topic,BUF_SIZE) ;

    strncpy(mqtt->identifier, identifier,BUF_SIZE) ;

    if(type == TYPE_PUB)

    {

        strncpy(mqtt->method, method,BUF_SIZE) ;

        strncpy(mqtt->json_id, json_id,BUF_SIZE) ;

        strncpy(mqtt->version, version,BUF_SIZE) ;

    }



#if 0

    printf(" MQTT hostname[%s]:%d,n clientid:%s,n username:%s, passwd:%s,n topic:%sn",

            mqtt->hostname, mqtt->port, mqtt->clientid, mqtt->username, mqtt->passwd, mqtt->topic);

#endif

    iniparser_freedict(ini); //Relaese ini file

    return 0 ;

}

一开始没改配置文件的话，可以通过传入参数去修改如果没有传入参数默认用配置文件中的配置：

可通过选项传参修改配置文件中的配置host,port,username,passwd,client_id,topic 这些参数

设置mqtt连接需要的参数：

复制代码

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
void set_mqtt_conf(char *ini_path,char *host,int port,char *id,char *user,char *passwd,char *topic)

{

    char    _mqtt_port[16] ;

    char    *mqtt_port ;



    if(port)

    {

        snprintf(_mqtt_port, sizeof(_mqtt_port), "%d", port) ;

        mqtt_port = _mqtt_port ;

    }

    else{

        mqtt_port = NULL ;

    }

    set_conf(ini_path,host,"mqtt_server_addr:host") ;

    set_conf(ini_path,mqtt_port,"mqtt_server_addr:port") ;

    set_conf(ini_path,id,"client_id:id") ;

    set_conf(ini_path,user,"user_passwd:username") ;

    set_conf(ini_path,passwd,"username:passwd") ;



    set_conf(ini_path,topic,"sub_topic:topic") ;

    set_conf(ini_path,topic,"pub_topic:topic") ;



}

设置到配置文件里面

复制代码

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
void set_conf(char *ini_path, char *value, const char * entry)

{

    int                 rv = 0;

    dictionary          *ini = NULL;

    FILE                *fp;

    

    if(!ini_path || !value || !entry)

    {

        return ;

    }

    printf("Ini file%s,set entry %s value=%sn",ini_path, entry, value) ;


    ini = iniparser_load(ini_path); //Load ini file get mqtt conf

    if ( ini == NULL ) {

        printf("iniparser_load error!n");

        return ;

    }



    if(iniparser_find_entry(ini,entry)){

        printf("the value is: %srn", iniparser_getstring(ini, entry, ""));

    }else{

        printf("the entry not exist...rn");

        return ;

    }



    rv = iniparser_set(ini, entry,value) ;

    if(rv != 0)

    {

        printf("iniparser_set %s=%s errorn",entry,value ) ;

        goto cleanup ;

    }

    fp = fopen(ini_path, "w");

    iniparser_dump_ini(ini, fp);

    fclose(fp);


cleanup:

    iniparser_freedict(ini); //Relaese ini file

}

mosq_sub的主函数

复制代码

int main(int argc, char **argv)

{

int                 rv = -1 ;

st_mqtt             mqtt ;

struct mosquitto    *mosq = NULL;

bool                session = true ;

st_mosq_ctx         mosq_ctx ;

int                 opt = -1 ;

int                 daemon_run = 0 ;

char                *program_name ;

char                *host=NULL ;

int                 port = 0 ;

char                *id = NULL ;

char                *user = NULL ;

char                *passwd = NULL ;

char                *topic = NULL ;

program_name = basename(argv[0]) ;

const char *short_opts = "i:h:p:P:u:t:dH";

const struct option long_opts[] =   {

{"host", required_argument, NULL, 'h'},

{"daemon", no_argument, NULL, 'd'},

{"help", no_argument, NULL, 'H'},

{"port", required_argument, NULL, 'p'},

{"id", required_argument, NULL, 'i'},

{"passwd", required_argument, NULL, 'P'},

{"user", required_argument, NULL, 'u'},

{"topic", required_argument, NULL, 't'},

{0, 0, 0, 0}

};

while ((opt= getopt_long(argc, argv, short_opts, long_opts,NULL)) != -1)

{

switch (opt)

{

case 'i':

id = optarg ;

break ;

case 'P':

passwd = optarg ;

break ;

case 'u':

user = optarg ;

break ;

case 'p':

port = atoi(optarg) ;

break ;

case 'd':

daemon_run = 1 ;

break ;

case 'h':

host = optarg ;

break ;

case 't':

topic  = optarg ;

break ;

case 'H':

print_usage(program_name) ;

return 0 ;

}

set_mqtt_conf(INI_PATH,host, port, id, user, passwd, topic) ;

memset(&mosq_ctx,0,sizeof(mosq_ctx)) ;

memset(&mqtt,0,sizeof(mqtt)) ;

rv = gain_mqtt_conf(INI_PATH,&mqtt,TYPE_PUB) ;

if(rv < 0)

{

printf("gain mqtt conf errorn") ;

return -1 ;

}

strncpy( mosq_ctx.topic, mqtt.topic, sizeof(mosq_ctx.topic)) ;

strncpy( mosq_ctx.identifier, mqtt.identifier, sizeof(mosq_ctx.identifier)) ;

/*  Program running in backgrund    */

if(daemon_run)

{

daemon(1,1) ;

}

/*   必须在任何其他mosquitto功能之前调用。 */

mosquitto_lib_init() ;

/*   创建一个新的mosquitto客户端实例 */

mosq = mosquitto_new(mqtt.clientid, session, (void*)&mosq_ctx) ;

if(!mosq)

{

printf("Mosquitto_new() failed: %sn", strerror(errno)) ;

goto cleanup ;

}

printf("Create mosquitto sucessfully!n");

/*      Set callback function    */

mosquitto_connect_callback_set(mosq, conn_callback) ;

mosquitto_message_callback_set(mosq, message_callback);

/*   配置mosquitton实例的用户名和密码    */

if( mosquitto_username_pw_set(mosq, mqtt.username, mqtt.passwd) != MOSQ_ERR_SUCCESS )

{

printf("Mosq_usee_pw_set() failed: %sn", strerror(errno) );

goto cleanup ;

}

if( mosquitto_connect(mosq, mqtt.hostname, mqtt.port, KEEP_ALIVE) )

{

printf("connect mqtt server failedn") ;

goto cleanup ;

}

mosquitto_loop_forever(mosq, -1, 1);

cleanup:

printf("program will exitn");

mosquitto_destroy(mosq) ;

mosquitto_lib_cleanup() ;

return 0 ;

}

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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
int main(int argc, char **argv)

{

    int                 rv = -1 ;

    st_mqtt             mqtt ;

    struct mosquitto    *mosq = NULL;

    bool                session = true ;

    st_mosq_ctx         mosq_ctx ;

    int                 opt = -1 ;

    int                 daemon_run = 0 ;

    char                *program_name ;

    char                *host=NULL ;

    int                 port = 0 ;

    char                *id = NULL ;

    char                *user = NULL ;

    char                *passwd = NULL ;

    char                *topic = NULL ;

    

    program_name = basename(argv[0]) ;



    const char *short_opts = "i:h:p:P:u:t:dH";                    

    const struct option long_opts[] =   {  

        {"host", required_argument, NULL, 'h'},    

        {"daemon", no_argument, NULL, 'd'},  

        {"help", no_argument, NULL, 'H'},  

        {"port", required_argument, NULL, 'p'},  

        {"id", required_argument, NULL, 'i'},  

        {"passwd", required_argument, NULL, 'P'},  

        {"user", required_argument, NULL, 'u'},  

        {"topic", required_argument, NULL, 't'},  

        {0, 0, 0, 0}

    };  

    while ((opt= getopt_long(argc, argv, short_opts, long_opts,NULL)) != -1)

    {

        switch (opt)

        {

            case 'i':

                id = optarg ;

                break ;

            case 'P':

                passwd = optarg ;

                break ;

            case 'u':

                user = optarg ;

                break ;

            case 'p':

                port = atoi(optarg) ;

                break ;

            case 'd':

                daemon_run = 1 ;

                break ;

            case 'h':

                host = optarg ;

                break ;

            case 't':

                topic  = optarg ;

                break ;

            case 'H':

                print_usage(program_name) ;

                return 0 ;

        }

    }

    set_mqtt_conf(INI_PATH,host, port, id, user, passwd, topic) ;


    memset(&mosq_ctx,0,sizeof(mosq_ctx)) ;

    memset(&mqtt,0,sizeof(mqtt)) ;


    rv = gain_mqtt_conf(INI_PATH,&mqtt,TYPE_PUB) ;

    if(rv < 0)

    {

        printf("gain mqtt conf errorn") ;

        return -1 ;

    }

    strncpy( mosq_ctx.topic, mqtt.topic, sizeof(mosq_ctx.topic)) ;

    strncpy( mosq_ctx.identifier, mqtt.identifier, sizeof(mosq_ctx.identifier)) ;



    /*  Program running in backgrund    */

    if(daemon_run)

    {

        daemon(1,1) ;

    }



    /*   必须在任何其他mosquitto功能之前调用。 */

    mosquitto_lib_init() ;


    /*   创建一个新的mosquitto客户端实例 */

    mosq = mosquitto_new(mqtt.clientid, session, (void*)&mosq_ctx) ;

    if(!mosq)

    {

        printf("Mosquitto_new() failed: %sn", strerror(errno)) ;

        goto cleanup ;

    }

    printf("Create mosquitto sucessfully!n");


    /*      Set callback function    */

    mosquitto_connect_callback_set(mosq, conn_callback) ;

    mosquitto_message_callback_set(mosq, message_callback);


    /*   配置mosquitton实例的用户名和密码    */

    if( mosquitto_username_pw_set(mosq, mqtt.username, mqtt.passwd) != MOSQ_ERR_SUCCESS )

    {

        printf("Mosq_usee_pw_set() failed: %sn", strerror(errno) );

        goto cleanup ;

    }



    if( mosquitto_connect(mosq, mqtt.hostname, mqtt.port, KEEP_ALIVE) )

    {

        printf("connect mqtt server failedn") ;

        goto cleanup ;

    }

    mosquitto_loop_forever(mosq, -1, 1);


cleanup:

    printf("program will exitn");

    mosquitto_destroy(mosq) ;

    mosquitto_lib_cleanup() ;



    return 0 ;

}

登录阿里云平台进行测试

我这里发布的是Data功能，使用阿里云平台的在线调试发布消息客户端收到消息

使用参数进行连接订阅

客户端收到的消息

下面一行打印的是通过配置里的[ali_json]：identifier = Data，标识符去找该标识符对应的值，也就是阿里云平台对应的功能。

发布端代码也大致如此就不再说了，还有存在BUG，如传参的时候设置用户名buntu&a1R4lbARNLO时遇到&后面的就报错a1R4lbARNLO：未找到命令，暂时还没解决，还有发布端一开始想着是可能会上报一些温度或者是某些需要用到硬件的获取的值，然后运行进行发布，载荷传参以及配置文件里都还没有....

在客户端mosq_pub.c代码中，可修改发布的内容和间隔时间 10秒钟发布一次