基于mosquitto库实现MQTT订阅与发布

Posted 2021-07-25 Shemesz

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了基于mosquitto库实现MQTT订阅与发布相关的知识，希望对你有一定的参考价值。

mosquitto库实现发布与订阅

一、下载安装mosquitto

（1）下载地址：https://mosquitto.org/download/
Linux命令行输入

wget https://mosquitto.org/files/source/mosquitto-1.6.12.tar.gz

（2）解压

tar -zxvf mosquitto-1.6.12.tar.gz

（3）编译安装

make
sudo make install

（4）解决依赖
在这里可能会遇到编译安装的问题，都是缺少相关依赖，提示错误信息里缺少什么，下载什么即可

sudo apt-get install libssl-dev
sudo apt-get install uuid-dev
sudo apt-get install cmake

二、mosquitto发布与订阅

发布端

#include <stdio.h>
#include <stdlib.h>
#include <unistd.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: my_connect_callback\\n");

}

void my_disconnect_callback(struct mosquitto *mosq, void *obj, int rc)
{
        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_set(mosq, my_connect_callback);
		mosquitto_disconnect_callback_set(mosq, my_disconnect_callback);
        mosquitto_publish_callback_set(mosq, my_publish_callback);

		/*连接至服务器 */
        /* 参数：句柄、ip（host）、端口、心跳 */
        ret = mosquitto_connect(mosq, HOST, PORT, KEEP_ALIVE);
        if(ret){
                printf("Connect server 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,"test",strlen(buff)+1,buff,0,0);
			memset(buff,0,sizeof(buff));
		}

        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        printf("End!\\n");

        return 0;
}

订阅端

#include <stdio.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);
        }else{
                /*订阅主题*/
                /* 参数：句柄、id、订阅的主题、qos*/
                if(mosquitto_subscribe(mosq, NULL, "test", 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);

        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_set(mosq, my_connect_callback);
		mosquitto_disconnect_callback_set(mosq, my_disconnect_callback);
        mosquitto_subscribe_callback_set(mosq, my_subscribe_callback);
        mosquitto_message_callback_set(mosq, my_message_callback);

        /*连接至服务器*/
        /*参数：句柄、ip（host）、端口、心跳*/
       ret = mosquitto_connect(mosq, HOST, PORT, KEEP_ALIVE);
        if(ret){
                printf("Connect server 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;
}

Makefile

all:
    @echo ""
    @echo "Start compiling......"
    @echo ""
    gcc -o sub sub.c -lmosquitto
    gcc -o pub pub.c -lmosquitto
    @echo "end"
sub:
    gcc -o sub sub.c -lmosquitto 

pub:
    gcc -o pub pub.c -lmosquitto 

clean:
    -rm sub pub