mqtt客户端演⽰（MQTT通信协议（mosquitto）发布订阅例⼦
C语⾔实现）
MQTT客户端
struct mosquitto *mosq;
void my_connect_callback(struct mosquitto *mosq, void *obj, int rc)
{
//printf("rc11: %d\n", rc);
}
void my_disconnect_callback(struct mosquitto *mosq, void *obj, int result)
{
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();
}
void my_publish_callback(struct mosquitto *mosq, void *obj, int mid)
{
//printf("%d ", message_count);
}
void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{
char *message = NULL;
message = (char *)msg->payload;
printf("recv len is: %d, message is %s\n", msg->payloadlen, msg->payload);
}
int mqttInit(void)
{
int mid = 0;
mosquitto_lib_init();
mosq = mosquitto_new("mqtt", true, NULL);
mosquitto_connect_callback_t(mosq, my_connect_callback);
mosquitto_disconnect_callback_t(mosq, my_disconnect_callback);
mosquitto_publish_callback_t(mosq, my_publish_callback);
mosquitto_message_callback_t(mosq, my_message_callback);
mosquitto_connect(mosq, "⾃⼰的 mqtt rver", 1883, 600);
mosquitto_subscribe(mosq, &mid, “sub的topic”, 1);
mosquitto_publish(mosq, NULL, “pub的topic“”, “发送的内容长度”, “所要发送的内容”, 1, fal);
return SUCCESSES;
}
int mqtt_run()
{
八条宫智仁亲王ret = pthread_create(&tid, &ta, (void *)mosquitto_loop_start, mosq);
}
{
镐的读音int n = 0;
while (n >= 0 && !miio.force_exit) {
int i;
n = poll(miio.pollfds, unt_pollfds, POLL_TIMEOUT);
if (n < 0) {
perror("poll");
continue;
}
if (n == 0) {
/* printf("poll timeout\n"); */
mosquitto_loop_sq);
continue;
}
for (i = 0; i < unt_pollfds && n > 0; i++) {
if (miio.pollfds[i].revents & POLLIN) {
if (miio.pollfds[i].fd == mosq_sock)
mosquitto_loop_sq, 1);
n--;
} el if (miio.pollfds[i].revents & POLLOUT) {
李秀成if (miio.pollfds[i].fd == mosq_sock)
mosquitto_loop_sq, 1);
n--;
}
}
}
}
输⼊输出轮询
那些仅仅是通过命令去测试，现在我们来通过mosquitto的官⽅源码，编程实现MQTT协议的发布订阅。
上⼀章中通过源码编译安装后，将会有⼏个我们需要⽤到的⽂件。对应路径如下：
mosquitto-1.6.10/lib/libmosquitto.so.1
mosquitto-1.6.10/lib/cpp/libmosquittopp.so.1
mosquitto-1.6.10/lib/mosquitto.h
mosquitto-1.6.10/lib/mosquittopp.h
下⾯的程序是使⽤标准C语⾔来实现，所以我们只需要⽤到mosquitto.h和libmosquitto.so.1两个⽂件就可以了。
同步函数
mosquitto有同步和异步两种通讯⽅式。这⾥的异步是⼀种⾮阻塞的⽅式，⽐同步通信性能更好，因为同步的⽅式是“通信+等待”的阻塞模式，不过接下来先编写⼀下mosquitto同步函数，再写异步函数。
具体函数如下：
如：mosquitto_connect
具体代码
订阅端（sub.c）
#include <stdio.h>
#include <stdlib.h>
#include <stdlib.h>
#include <string.h>
#include "mosquitto.h"
#define HOST "localhost"
#define PORT 1883
#define KEEP_ALIVE 60
#define MSG_MAX_SIZE 512
// 定义运⾏标志决定是否需要结束
static int running = 1;
void my_connect_callback(struct mosquitto *mosq, void *obj, int rc)
{
printf("Call the function: on_connect\n");
if(rc){
// 连接错误，退出程序
printf("on_connect error!\n");
exit(1);
}el{
// 订阅主题流行音乐
// 参数：句柄、id、订阅的主题、qos
if(mosquitto_subscribe(mosq, NULL, "topic1", 2)){
printf("Set the topic error!\n");
exit(1);
}
}
}
void my_disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
{
printf("Call the function: my_disconnect_callback\n");
running = 0;
}
void my_subscribe_callback(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos) {
printf("Call the function: on_subscribe\n");
}
void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{
printf("Call the function: on_message\n");
printf("Recieve a message of %s : %s\n", (char *)msg->topic, (char *)msg->payload);bamboo
if(0 == strcmp(msg->payload, "quit")){
mosquitto_disconnect(mosq);
}
}
int main()
{
int ret;
struct mosquitto *mosq;
// 初始化mosquitto库
ret = mosquitto_lib_init();
if(ret){
printf("Init lib error!\n");
return -1;
}
// 创建⼀个订阅端实例
// 参数：id（不需要则为NULL）、clean_start、⽤户数据
mosq = mosquitto_new("sub_test", true, NULL);
if(mosq == NULL){
printf("New sub_test error!\n");
mosquitto_lib_cleanup();
return -1;
}
// 设置回调函数
// 参数：句柄、回调函数
mosquitto_connect_callback_t(mosq, my_connect_callback);
mosquitto_disconnect_callback_t(mosq, my_disconnect_callback); mosquitto_subscribe_callback_t(mosq, my_subscribe_callback); mosquitto_message_callback_t(mosq, my_message_callback);
// 连接⾄服务器
// 参数：句柄、ip（host）、端⼝、⼼跳
ret = mosquitto_connect(mosq, HOST, PORT, KEEP_ALIVE);
圣湖if(ret){
printf("Connect rver error!\n");
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();
return -1;
}
// 开始通信：循环执⾏、直到运⾏标志running被改变
printf("Start!\n");
while(running)
{
键盘数字键mosquitto_loop(mosq, -1, 1);
}
// 结束后的清理⼯作
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();
printf("End!\n");
return 0;
}
发布端（pub.c）
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include "mosquitto.h"
#define HOST "localhost"
#define PORT 1883
#define KEEP_ALIVE 60
#define MSG_MAX_SIZE 512
static int running = 1;
void my_connect_callback(struct mosquitto *mosq, void *obj, int rc)
{
printf("Call the function: my_connect_callback\n");
}
void my_disconnect_callback(struct mosquitto *mosq, void *obj, int rc) {
printf("Call the function: my_disconnect_callback\n");
printf("Call the function: my_disconnect_callback\n");
running = 0;
}
void my_publish_callback(struct mosquitto *mosq, void *obj, int mid)
{
printf("Call the function: my_publish_callback\n");
}
int main()
{
int ret;
struct mosquitto *mosq;
char buff[MSG_MAX_SIZE];
//初始化libmosquitto库
ret = mosquitto_lib_init();
if(ret){
printf("Init lib error!\n");
读书的方法return -1;
}
//创建⼀个发布端实例
mosq = mosquitto_new("pub_test", true, NULL);
if(mosq == NULL){
printf("New pub_test error!\n");
mosquitto_lib_cleanup();
return -1;
}
//设置回调函数
mosquitto_connect_callback_t(mosq, my_connect_callback);
mosquitto_disconnect_callback_t(mosq, my_disconnect_callback);
mosquitto_publish_callback_t(mosq, my_publish_callback);
// 连接⾄服务器
/
/ 参数：句柄、ip（host）、端⼝、⼼跳
ret = mosquitto_connect(mosq, HOST, PORT, KEEP_ALIVE);
if(ret){
printf("Connect rver error!\n");
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();
return -1;
}
printf("Start!\n");
//mosquitto_loop_start作⽤是开启⼀个线程，在线程⾥不停的调⽤ mosquitto_loop() 来处理⽹络信息
int loop = mosquitto_loop_start(mosq);
if(loop != MOSQ_ERR_SUCCESS)
{
printf("mosquitto loop error\n");
return 1;
}
while(fgets(buff, MSG_MAX_SIZE, stdin) != NULL)
{
/*发布消息*/
mosquitto_publish(mosq,NULL,"topic1",strlen(buff)+1,buff,0,0);
memt(buff,0,sizeof(buff));
}
mosquitto_destroy(mosq);
