mqtt应用于进程间通信实例解析丨【拜托了，物联网！】

举报
嵌入式大杂烩 发表于 2021/10/15 23:06:19 2021/10/15
【摘要】 mqtt应用于进程间通信实例解析

前言

上一篇分享了：《简单认识认识mqtt及mosquitto》，但也只是分享了mqtt的一些概念及mosquitto的一些介绍。这不，就有读者来催更了：

=======000

这一篇我们就来分享mqtt应用于进程间通信的实例。我们沿用《例说嵌入式实用知识之JSON数据》这篇文章的综合demo来改造改造。那个综合demo的功能是这样子的：

======001

这是以socket来作为进程间通信的方式，并且这个demo是基于Windows写的，需要包含Windows特定的头文件。

本篇笔记我们把上面这个综合demo改为：

======002

我们用mqtt来作为进程间通信的方式，在Linux下进程测试。

先贴代码：

json_print进程源码

json_print.c：

/*
- 程序功能: 组JSON格式数据包并发送(MQTT发布者客户端程序)
- 编译命令: gcc cJSON.c json_print.c -L ../mosquitto/build/lib -lmosquitto -o json_print
- 导出mosquitto动态库: export LD_LIBRARY_PATH=../mosquitto/build/lib:$LD_LIBRARY_PATH
- 作者：ZhengN
- 公众号：嵌入式大杂烩
*/
​
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "cJSON.h"
#include "../mosquitto/lib/mosquitto.h"
​
#define  STU_NAME_LEN  32
​
/* 学生结构体 */
typedef struct _Student
{
    char name[STU_NAME_LEN];  // 名字      
    int num;                  // 学号      
    int c_score;              // C语言分数
}StudentDef, *pStudentDef;
​
/* 内部函数声明 */
static StudentDef StudentData_Prepare(void);
static char *StudentsData_Packet(pStudentDef _Stu);
static void StudentData_Send(const char *_data);
static void MqttClientInit(void);
static void MqttClientClean(void);
​
bool clean_session = true;
struct mosquitto *mosq = NULL;
​
/********************************************************************************************************
** 函数: main
**------------------------------------------------------------------------------------------------------
** 参数: 
** 说明: 
** 返回: 
********************************************************************************************************/
int main(void)
{
    StudentDef stu = {0};
    char *stu_data = NULL;
    int stu_count = 0;
    int i = 0;
​
    /* MQTT客户端初始化 */
    MqttClientInit();
​
    /* 需要登记的学生总人数 */
    printf("Please input number of student: ");
    scanf("%d", &stu_count);
​
    while (i++ < stu_count)
    {
        /* 准备数据 */
        stu = StudentData_Prepare();
​
        /* JSON格式数据组包 */
        stu_data = StudentsData_Packet(&stu);
​
        /* 发送数据 */
        StudentData_Send(stu_data);
    }
​
    /* 回收操作 */
    MqttClientClean();
​
    return 0;
}
​
/********************************************************************************************************
** 函数: StudentData_Prepare, 准备组包需要的数据
**------------------------------------------------------------------------------------------------------
** 参数: 
** 说明: 
** 返回: 获取得到的数据
********************************************************************************************************/
static StudentDef StudentData_Prepare(void)
{
    char name[STU_NAME_LEN] = {0};
    int num = 0;
    int c_score = 0;
    StudentDef stu;
​
    /* 名字 */
    printf("Please input name: ");
    scanf("%s", name);
    if (strlen(name) < STU_NAME_LEN)
    {
        strncpy((char*)&stu.name, name, strlen(name)+1);
    }
    else
    {
        printf("The name is too long\n");
    }
    
    /* 学号 */
    printf("Please input num (0~100): ");
    scanf("%d", &num);
    stu.num = num;
​
    /* C语言分数 */
    printf("Please input c_score (0~100): ");
    scanf("%d", &c_score);
    stu.c_score = c_score;
​
    return stu;
}
​
/********************************************************************************************************
** 函数: StudentsData_Packet, JSON格式数据组包
**------------------------------------------------------------------------------------------------------
** 参数: _Stu：组student json数据包需要的数据
** 说明: 
** 返回: JSON格式的字符串   
********************************************************************************************************/
static char *StudentsData_Packet(pStudentDef _Stu)
{
    char *res_string = NULL;    // 返回值
    cJSON *name = NULL;         // 名字
    cJSON *num = NULL;          // 学号
    cJSON *c_score = NULL;      // C语言分数
​
    /* 创建一个JSON对象，｛｝扩起来 */
    cJSON *obj = cJSON_CreateObject();
    if (obj == NULL)
    {
        goto end;
    }
​
    /* 创建 "name": "xxx" 键值对 */
    name = cJSON_CreateString(_Stu->name);
    if (name == NULL)
    {
        goto end;
    }
    cJSON_AddItemToObject(obj, "name", name);
​
    /* 创建 "num": 207 键值对 */
    num = cJSON_CreateNumber(_Stu->num);
    if (name == NULL)
    {
        goto end;
    }
    cJSON_AddItemToObject(obj, "num", num);
    
    /* 创建 "c_score": 95 键值对 */
    c_score = cJSON_CreateNumber(_Stu->c_score);
    if (name == NULL)
    {
        goto end;
    }
    cJSON_AddItemToObject(obj, "c_score", c_score); 
​
    res_string = cJSON_Print(obj);          // 呈现为JSON格式 
    // res_string = cJSON_PrintUnformatted(obj);   // 呈现为无格式
​
    if (res_string == NULL)
    {
        fprintf(stderr, "Failed to print monitor.\n");
    }
​
/* 异常情况统一Delete（free） */
end:
    cJSON_Delete(obj);
    return res_string;
}
​
/********************************************************************************************************
** 函数: StudentData_Send, JSON格式字符串数据组包发送
**------------------------------------------------------------------------------------------------------
** 参数: _data：要发送的数据
** 说明: 
** 返回: JSON格式的字符串   
********************************************************************************************************/
static void StudentData_Send(const char *_data)
{
    printf("%s: %s\n\n", __FUNCTION__, _data);
​
    /* 发布消息 */
    mosquitto_publish(mosq, NULL, "test_topic", strlen(_data)+1, (const char*)_data, 0, 0);
}
​
/********************************************************************************************************
** 函数: MqttClientInit, MQTT客户端初始化
**------------------------------------------------------------------------------------------------------
** 参数: void
** 说明: 
** 返回: 
********************************************************************************************************/
static void MqttClientInit(void)
{
    /* libmosquitto 库初始化 */
    mosquitto_lib_init();
​
    /* 创建mosquitto客户端 */
    mosq = mosquitto_new(NULL, clean_session, NULL);
    if(NULL == mosq)
    {
        printf("Create mqtt client failed...\n");
        mosquitto_lib_cleanup();
        return;
    }
​
    /* 连接服务器 */
    if(mosquitto_connect(mosq, "localhost", 1883, 60))
    {
        printf("Unable to connect...\n");
        return;
    }
​
    /* 网络消息处理线程 */
    int loop = mosquitto_loop_start(mosq);
    if(loop != MOSQ_ERR_SUCCESS)
    {
        printf("mosquitto loop error\n");
        return;
    }
}
​
/********************************************************************************************************
** 函数: MqttClientClean, MQTT客户端清理操作
**------------------------------------------------------------------------------------------------------
** 参数: void
** 说明: 
** 返回: 
********************************************************************************************************/
static void MqttClientClean(void)
{
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
}

json_parse进程源码

json_parse.c：

/*
- 程序功能: 接收JSON数据并解析(MQTT订阅者客户端程序)
- 编译命令: gcc cJSON.c json_parse.c -L ../mosquitto/build/lib -lmosquitto -o json_parse
- 导出mosquitto动态库: export LD_LIBRARY_PATH=../mosquitto/build/lib:$LD_LIBRARY_PATH
- 作者：ZhengN
- 公众号：嵌入式大杂烩
*/
​
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "cJSON.h"
#include "../mosquitto/lib/mosquitto.h"
​
#define  STU_NAME_LEN  32
​
/* 学生结构体 */
typedef struct _Student
{
    char name[STU_NAME_LEN];  // 名字      
    int num;                  // 学号      
    int c_score;              // C语言分数
}StudentDef, *pStudentDef;
​
/* 内部函数声明 */
static void StudentsData_Parse(pStudentDef _Stu, const char *_JsonStudnetData);
static void PrintParseResult(const pStudentDef _Stu);
static void SaveParseResult(const pStudentDef _Stu);
static void my_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message);
static void my_connect_callback(struct mosquitto *mosq, void *userdata, int result);
​
/* 内部全局变量 */
static FILE *stu_fp = NULL;
​
/********************************************************************************************************
** 函数: main
**------------------------------------------------------------------------------------------------------
** 参数: 
** 说明: 
** 返回: 
********************************************************************************************************/
bool clean_session = true;
int main(void)
{         
    struct mosquitto *mosq = NULL;
​
    /* libmosquitto 库初始化 */
    mosquitto_lib_init();
​
    /* 创建mosquitto客户端 */
    mosq = mosquitto_new(NULL, clean_session, NULL);
    if(NULL == mosq)
    {
        printf("Create mqtt client failed...\n");
        mosquitto_lib_cleanup();
        return 1;
    }
​
    /* 绑定连接、消息接收回调函数 */
    mosquitto_connect_callback_set(mosq, my_connect_callback);
    mosquitto_message_callback_set(mosq, my_message_callback);
​
    /* 连接服务器 */
    if(mosquitto_connect(mosq, "localhost", 1883, 60))
    {
        printf("Unable to connect...\n");
        return 1;
    }
​
    /* 循环处理网络消息 */
    mosquitto_loop_forever(mosq, -1, 1);
​
    /* 回收操作 */
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
​
    return 0;
}
​
/********************************************************************************************************
** 函数: StudentsData_Parse, JOSN格式学生期末数据解析
**------------------------------------------------------------------------------------------------------
** 参数: _JsonStudnetData：JSON数据   _Stu：保存解析出的有用数据
** 说明: 
** 返回: 
********************************************************************************************************/
static void StudentsData_Parse(pStudentDef _Stu, const char *_JsonStudnetData)
{
    cJSON *student_json = NULL;   // student_json操作对象，可代表 {} 扩起来的内容
    cJSON *name = NULL;             
    cJSON *num = NULL;
    cJSON *c_score = NULL;
    
    /* 开始解析 */
    student_json = cJSON_Parse(_JsonStudnetData);
    if (NULL == student_json)
    {
        const char *error_ptr = cJSON_GetErrorPtr();
        if (error_ptr != NULL)
        {
            fprintf(stderr, "Error before: %s\n", error_ptr);
        }
        goto end;
    }
​
    /* 解析获取name得值 */
    name = cJSON_GetObjectItemCaseSensitive(student_json, "name");
    if (cJSON_IsString(name) && (name->valuestring != NULL))
    {
        memset(&_Stu->name, 0, STU_NAME_LEN*sizeof(char));
        memcpy(&_Stu->name, name->valuestring, strlen(name->valuestring));
    }
​
    /* 解析获取num的值 */
    num = cJSON_GetObjectItemCaseSensitive(student_json, "num");
    if (cJSON_IsNumber(num))
    {
        _Stu->num = num->valueint;
    }
​
    /* 解析获取c_score的值 */
    c_score = cJSON_GetObjectItemCaseSensitive(student_json, "c_score");
    if (cJSON_IsNumber(c_score))
    {
        _Stu->c_score = c_score->valueint;
    }
​
end:
    cJSON_Delete(student_json);
}
​
/********************************************************************************************************
** 函数: PrintParseResult, 打印输出解析结果
**------------------------------------------------------------------------------------------------------
** 参数: 
** 说明: 
** 返回: 
********************************************************************************************************/
static void PrintParseResult(const pStudentDef _Stu)
{
    printf("name: %s, num: %d, c_score: %d\n\n", _Stu->name, _Stu->num, _Stu->c_score);
}
​
/********************************************************************************************************
** 函数: SaveParseResult, 保存解析结果
**------------------------------------------------------------------------------------------------------
** 参数: _Stu：需要保存的数据
** 说明: 
** 返回: 
********************************************************************************************************/
static void SaveParseResult(const pStudentDef _Stu)
{
    char write_buf[512] = {0};
    static int stu_count = 0;
​
    /* 以可在文件末尾追加内容的方式打开文件 */
    if((stu_fp = fopen("ParseResult.txt", "a+")) == NULL)
    {
        printf("Open file error!\n");
        return exit(EXIT_FAILURE);
    } 
​
    /* 按指定格式写入文件 */
    snprintf(write_buf, 512, "name: %s, num: %d, c_score: %d\n", _Stu->name, _Stu->num, _Stu->c_score);
    size_t len = fwrite((char*)write_buf, 1, strlen(write_buf), stu_fp);
​
    /* 文件位置指针偏移 */
    fseek(stu_fp, len * stu_count, SEEK_SET);
    stu_count++;
​
    /* 关闭文件 */
    fclose(stu_fp);
}
​
/********************************************************************************************************
** 函数: my_message_callback, 消息接收回调函数
**------------------------------------------------------------------------------------------------------
** 参数: 
** 返回: 
********************************************************************************************************/
static void my_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
{
    StudentDef stu = {0};
​
    if(message->payloadlen)
    {
        printf("%s %s\n", message->topic, (char*)message->payload);
​
        /* 解析JSON数据 */
        StudentsData_Parse(&stu, (const char*)message->payload);  
​
        /* 打印输出解析结果 */
        PrintParseResult(&stu);  
​
        /* 保存数据到文件 */
        SaveParseResult(&stu); 
    }
    else
    {
        printf("%s (null)\n", message->topic);
    }
    fflush(stdout);
}
​
/********************************************************************************************************
** 函数: my_connect_callback, 连接回调函数
**------------------------------------------------------------------------------------------------------
** 参数: 
** 返回: 
********************************************************************************************************/
static void my_connect_callback(struct mosquitto *mosq, void *userdata, int result)
{
    if(!result)
    {
        /* 订阅test_topic主题的消息 */
        mosquitto_subscribe(mosq, NULL, "test_topic", 0);
    }
    else
    {
        fprintf(stderr, "Connect failed\n");
    }
}

编译运行

1、编译生成json_parse、json_print程序：

gcc cJSON.c json_parse.c -L ../mosquitto/build/lib -lmosquitto -o json_parse
gcc cJSON.c json_print.c -L ../mosquitto/build/lib -lmosquitto -o json_print

这里用到链接动态库的方式生成可执行程序。关于动态链接与静态链接，可查看往期笔记：《静态链接与动态链接补充（Linux）》、《什么是动态链接与静态链接？》。

2、执行json_parse、json_print程序：

执行这两个程序会报错：

======003

./json_parse: error while loading shared libraries: libmosquitto.so.1: cannot open shared object file: No such file or directory
/json_print: error while loading shared libraries: libmosquitto.so.1: cannot open shared object file: No such file or directory

这是因为 不能找到共享库文件libmosquitto.so.1，加载失败。

因为一般情况下Linux会在/usr/lib路径中搜索需要用到的库，而libmosquitto.so.1库并不在这个路径下。

解决方法有两种：一种就是把这个文件拷贝至/usr/lib路径下，但是一般不允许这样做，一般用户也不允许往这个路径里拷贝东西。另一种就是把libmosquitto.so.1库所在路径增加为动态库的搜索路径，命令为：

export LD_LIBRARY_PATH=../mosquitto/build/lib:$LD_LIBRARY_PATH

关于这方面的说明可以阅读往期笔记：《静态链接与动态链接补充（Linux）》

按照上述方法添加动态库搜索路径之后就可以正常运行这两个程序：

======004

ParseResult.txt文本里得到：

======005

实验成功！

以上就是本次的分享，代码写得比较仓促，如有错误，麻烦指出，谢谢！由于准备demo花了挺多时间，包括注释也写了很多。所以本篇文章就不做过多的说明，感兴趣的朋友可以结合本篇文章的demo及mosquitto/client/pub_client.c、mosquitto/client/sub_client.c这两个源文件。

本篇文章的demo：

=======006

可在本公众号聊天界面回复关键词：json_mqtt_demo，即可获取，若无法获取可联系我进行获取。

【拜托了，物联网！】有奖征文火热进行中：https://bbs.huaweicloud.cn/blogs/299476

【版权声明】本文为华为云社区用户原创内容，转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息， 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com
IoT MQTT 任务调度
  • 点赞
  • 收藏
  • 关注作者

评论（0）

0/1000
抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认 取消

加入云驻计划，成为创作者

  • 华为云周边好礼
  • 免费体验产品
  • 特殊身份标识
  • 线下官方门票
  • 内部专家零距离
  • 与10000+优质创作者共同成长
立即加入