可订阅可发布模式

具体代码

客户端1代码：pub.c

#include <stdio.h>
#include <stdlib.h>
#include <mosquitto.h>
#include <string.h>
 
#define HOST "localhost"
#define PORT  1883
#define KEEP_ALIVE 60
#define MSG_MAX_SIZE  512
 
bool session = true;
 
void my_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
{
	
    if(message->payloadlen){
        printf("%s %s", message->topic, (char *)message->payload);
    }else{
        printf("%s (null)\n", message->topic);
    }
    fflush(stdout);
}
 
void my_connect_callback(struct mosquitto *mosq, void *userdata, int result)
{
    int i;
    if(!result){
        /* Subscribe to broker information topics on successful connect. */
        mosquitto_subscribe(mosq, NULL, "topic2 ", 2);
    }else{
        fprintf(stderr, "Connect failed\n");
    }
}
 
void my_subscribe_callback(struct mosquitto *mosq, void *userdata, int mid, int qos_count, const int *granted_qos)
{
    int i;
    printf("Subscribed (mid: %d): %d", mid, granted_qos[0]);
    for(i=1; i<qos_count; i++){
        printf(", %d", granted_qos[i]);
    }
    printf("\n");
}
 
void my_log_callback(struct mosquitto *mosq, void *userdata, int level, const char *str)
{
    printf("%s\n", str);
}
 
int main()
{
    struct mosquitto *mosq = NULL;
	char buff[MSG_MAX_SIZE];
	
    //libmosquitto 库初始化
    mosquitto_lib_init();
    //创建mosquitto客户端
    mosq = mosquitto_new(NULL,session,NULL);
    if(!mosq){
        printf("create client failed..\n");
        mosquitto_lib_cleanup();
        return 1;
    }
    //设置回调函数，需要时可使用
    mosquitto_log_callback_set(mosq, my_log_callback);
    mosquitto_connect_callback_set(mosq, my_connect_callback);
    mosquitto_message_callback_set(mosq, my_message_callback);
    mosquitto_subscribe_callback_set(mosq, my_subscribe_callback);
	
 
    //连接服务器
    if(mosquitto_connect(mosq, HOST, PORT, KEEP_ALIVE)){
        fprintf(stderr, "Unable to connect.\n");
        return 1;
    }
    //开启一个线程，在线程里不停的调用 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);
        memset(buff,0,sizeof(buff));
    }
 
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
 
    return 0;
}

客户端2代码：sub.c 

#include <stdio.h>
#include <stdlib.h>
#include <mosquitto.h>
#include <string.h>
 
#define HOST "localhost"
#define PORT  1883
#define KEEP_ALIVE 60
#define MSG_MAX_SIZE  512
#define TOPIC_NUM 3
 
bool session = true;
 
void my_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
{
	
    if(message->payloadlen){
        printf("%s %s", message->topic, (char *)message->payload);
    }else{
        printf("%s (null)\n", message->topic);
    }
    fflush(stdout);
}
 
void my_connect_callback(struct mosquitto *mosq, void *userdata, int result)
{
    int i;
    if(!result){
        /* Subscribe to broker information topics on successful connect. */
        mosquitto_subscribe(mosq, NULL, "topic1 ", 2);
    }else{
        fprintf(stderr, "Connect failed\n");
    }
}
 
void my_subscribe_callback(struct mosquitto *mosq, void *userdata, int mid, int qos_count, const int *granted_qos)
{
    int i;
    printf("Subscribed (mid: %d): %d", mid, granted_qos[0]);
    for(i=1; i<qos_count; i++){
        printf(", %d", granted_qos[i]);
    }
    printf("\n");
}
 
void my_log_callback(struct mosquitto *mosq, void *userdata, int level, const char *str)
{
    /* Pring all log messages regardless of level. */
    printf("%s\n", str);
}
 
int main()
{
    struct mosquitto *mosq = NULL;
	char buff[MSG_MAX_SIZE];
	
    //libmosquitto 库初始化
    mosquitto_lib_init();
    //创建mosquitto客户端
    mosq = mosquitto_new(NULL,session,NULL);
    if(!mosq){
        printf("create client failed..\n");
        mosquitto_lib_cleanup();
        return 1;
    }
    //设置回调函数，需要时可使用
    mosquitto_log_callback_set(mosq, my_log_callback);
    mosquitto_connect_callback_set(mosq, my_connect_callback);
    mosquitto_message_callback_set(mosq, my_message_callback);
    mosquitto_subscribe_callback_set(mosq, my_subscribe_callback);
	
 
    //连接服务器
    if(mosquitto_connect(mosq, HOST, PORT, KEEP_ALIVE)){
        fprintf(stderr, "Unable to connect.\n");
        return 1;
    }
    //开启一个线程，在线程里不停的调用 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,"topic2 ",strlen(buff)+1,buff,0,0);
        memset(buff,0,sizeof(buff));
    }
 
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
 
    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

运行结果

1、先执行make
2、启动mosquitto服务器
3、运行pub和sub
4、测试发布订阅信息,每个客户端都可以发布和订阅，双向收发。