C语言实现设计模式—订阅者/发布者模式（Pub-Sub Pattern）

2023-10-11 14:40
文章标签 语言 实现 设计模式 模式 订阅 pattern sub pub 发布者

本文主要是介绍C语言实现设计模式—订阅者/发布者模式（Pub-Sub Pattern），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

  • C语言实现订阅者/发布者模式（Pub-Sub Pattern）
    • 模式介绍
    • 模式的优点
    • 订阅者/发布者模式实现
      • 节点定义
      • 发布者/订阅者接口注册
      • 统一调度实现
      • 外层接口

C语言实现订阅者/发布者模式（Pub-Sub Pattern）

模式介绍

发布订阅模式中，称为发布者的消息发送者不会将消息直接发送给订阅者，这意味着发布者和订阅者不知道彼此的存在。在发布者和订阅者之间存在第三个组件，称为消息代理或调度中心或中间件，它维持着发布者和订阅者之间的联系，过滤所有发布者传入的消息并相应地分发它们给订阅者。

订阅者把自己想订阅的事件注册到调度中心，当该事件触发时候，发布者发布该事件到调度中心，由调度中心统一调度订阅者注册到调度中心的处理代码。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qJwesHLW-1597830017185)(C:\Users\Administrator\AppData\Roaming\Typora\typora-user-images\image-20200819170525741.png)]

模式的优点

  1. 发布者与订阅者松耦合，甚至不需要知道它们的存在。由于主题才是关注的焦点，发布者和订阅者可以对系统拓扑结构保持一无所知。各自继续正常操作而无需顾及对方。

  2. 不需要关心不同的组件是如何组合在一起的，只要他们共同遵守一份协议即可。

  3. 异步的消息传递有助于应用程序在增加的负载下继续平稳运行，并且可以更有效地处理间歇性故障。发布者/订阅者模式更像是一种跨应用程序模式。发布服务器和订阅服务器可以驻留在两个不同的应用程序中。它们中的每一个都通过消息代理或消息队列进行通信。

订阅者/发布者模式实现

节点定义

#ifndef __MQ_MAGIC_H__
#define __MQ_MAGIC_H__#include <stdbool.h>
#include "lw_oopc.h"
#include <rtthread.h>typedef int32_t topic_t;typedef enum user_mode_def{USER_MODE_PUB,            /* Publisher */USER_MODE_SUB             /* Subscribe */  
}user_mode;typedef struct mq_magic_frame_def{topic_t head;int32_t value;
}mq_magic_frame_t, pmq_magic_frame_t;typedef struct mq_magic_def{rt_list_t   user_list;   rt_mq_t     rt_mq;user_mode   user;topic_t     topic;
}mq_magic_t, *pmq_magic_t;typedef mq_magic_t pub_mq_magic_t;
typedef mq_magic_t sub_mq_magic_t;typedef struct rz_mq_magic_t rz_mq_magic_t;CLASS(rz_mq_magic_t)
{rt_device_t tid_device; rt_list_t   pub_list;rt_list_t   sub_list;/* 发布者注册接口 */bool (*publisher_register)(rz_mq_magic_t *t, mq_magic_t *pmq, topic_t topic);/* 订阅者注册接口 */bool (*subscriber_register)(rz_mq_magic_t *t, mq_magic_t *pmq, topic_t topic);void (*init)(rz_mq_magic_t *t);};rz_mq_magic_t *rz_mq_magic_obj_get(void);
bool mq_magic_user_create(mq_magic_t *pmq, const char *name, uint8_t flag);
rt_err_t rz_mq_magic_send(mq_magic_t *t, mq_magic_frame_t *buffer);
rt_err_t rz_mq_magic_recv(mq_magic_t *t, void *buffer, int32_t timeout);
rt_err_t rz_mq_magic_send_wait_receive(pub_mq_magic_t *ppub, sub_mq_magic_t *psub, mq_magic_frame_t *buffer, int32_t timeout);
int rz_mq_magic_init(void);
#endif

发布者/订阅者接口注册

static pub_mq_magic_t *_find_publisher(rz_mq_magic_t *t, pub_mq_magic_t *pmq)
{rt_list_t *pos;pub_mq_magic_t *ppub;rt_list_for_each(pos, &t->pub_list) {ppub = (pub_mq_magic_t *)rt_container_of(pos, pub_mq_magic_t, user_list);if(ppub == pmq) {return ppub;}}return NULL;}static sub_mq_magic_t *_find_subscriber(rz_mq_magic_t *t, sub_mq_magic_t *pmq)
{rt_list_t *pos;pub_mq_magic_t *psub;rt_list_for_each(pos, &t->sub_list) {psub = (sub_mq_magic_t *)rt_container_of(pos, sub_mq_magic_t, user_list);if(psub == pmq) {return psub;}}return NULL;}static bool publisher_register(rz_mq_magic_t *t, mq_magic_t *pmq, topic_t topic)
{if (NULL == pmq) {rt_kprintf("%s():(NULL == pmq) invalid.\r\n", __FUNCTION__);return false;}/* _find */if (NULL != _find_publisher(t, (pub_mq_magic_t *)pmq)) {rt_kprintf("%s():don't register (%x) repeated.\r\n", __FUNCTION__, pmq);return false;}pmq->user = USER_MODE_PUB;pmq->topic = topic;/* insert list */rt_list_insert_after(&t->pub_list, &pmq->user_list);return true;}static bool subscriber_register(rz_mq_magic_t *t, mq_magic_t *pmq, topic_t topic)
{if (NULL == pmq) {rt_kprintf("%s():(NULL == pmq) invalid.\r\n", __FUNCTION__);return false;}/* _find */if (NULL != _find_subscriber(t, (sub_mq_magic_t *)pmq)) {rt_kprintf("%s():don't register (%x) repeated.\r\n", __FUNCTION__, pmq);return false;}pmq->user = USER_MODE_SUB;pmq->topic = topic;/* insert list */rt_list_insert_after(&t->sub_list, &pmq->user_list);return true;}

统一调度实现

#define RTHREAD_PRIORITY_MQ_MAGIC         28
#define THREAD_STACK_SIZE_MQ_MAGIC        1024
#define THREAD_TIMESLICE_MQ_MAGIC         5
static rz_mq_magic_t *rz_mq_magic_obj = NULL;/* use singleton pattern */
rz_mq_magic_t *rz_mq_magic_obj_get(void)
{if (NULL == rz_mq_magic_obj) {rz_mq_magic_obj = rz_mq_magic_t_new();if (NULL == rz_mq_magic_obj) {rt_kprintf("rz_mq_magic_t_new failed\r\n");}}return (rz_mq_magic_t *)rz_mq_magic_obj;
}/* receive meassage, then return topic */
topic_t mq_magic_recv(mq_magic_t *t, mq_magic_frame_t *buff, int32_t timeout)
{if (RT_EOK == rt_mq_recv(t->rt_mq, buff, sizeof(mq_magic_frame_t), timeout)) {return t->topic;}return 0;
}/* base to topic, push meassage to ubscriber who subscibe this topic */
rt_err_t mq_magic_send(rz_mq_magic_t *t, topic_t topic, mq_magic_frame_t *buff)
{rt_list_t *pos;mq_magic_t *pmq;bool is_success = false;rt_list_for_each(pos, &t->sub_list) {pmq = (sub_mq_magic_t *)rt_container_of(pos, sub_mq_magic_t, user_list);if (topic == pmq->topic) {if (RT_EOK != rt_mq_send(pmq->rt_mq, buff, sizeof(mq_magic_frame_t))) {continue;}is_success = true;}}if (!is_success) {return RT_ERROR;}return RT_EOK;}/* main work : message relay service */
static void mq_magic_broker_process(void *parameter)
{rz_mq_magic_t *t = (rz_mq_magic_t *)parameter;rt_list_t *pos;mq_magic_t *pmq;mq_magic_frame_t mq_magic_data;topic_t topic;while(1) {/* check all publishers */rt_list_for_each(pos, &t->pub_list) {pmq = (pub_mq_magic_t *)rt_container_of(pos, pub_mq_magic_t, user_list);topic = mq_magic_recv(pmq, &mq_magic_data, RT_WAITING_NO);if (topic > 0) {mq_magic_send(t, topic, &mq_magic_data);/* send finish, reset topic */topic = 0;memset(&mq_magic_data, 0, sizeof(mq_magic_data));}}rt_thread_delay(10);}}static void mq_magic_init(rz_mq_magic_t *t)
{rt_list_init(&t->pub_list);rt_list_init(&t->sub_list);rt_thread_t tid  = RT_NULL;tid = rt_thread_create("mq_magic",mq_magic_broker_process, t,THREAD_STACK_SIZE_MQ_MAGIC,RTHREAD_PRIORITY_MQ_MAGIC,THREAD_TIMESLICE_MQ_MAGIC);if (RT_NULL != tid) {rt_thread_startup(tid);}}CTOR(rz_mq_magic_t)
FUNCTION_SETTING(publisher_register, publisher_register);
FUNCTION_SETTING(subscriber_register, subscriber_register);
FUNCTION_SETTING(init, mq_magic_init);
END_CTOR

外层接口

/* ---------------------------------------------- */
bool mq_magic_user_create(mq_magic_t *pmq, const char *name, uint8_t flag)
{if (NULL == pmq) {rt_kprintf("%s():(NULL == pmq) invalid.\r\n", __FUNCTION__);return false;}/* create rt_mq */pmq->rt_mq = rt_mq_create(name, sizeof(mq_magic_frame_t), 10, flag);if (NULL == pmq->rt_mq) {rt_kprintf("%s():rt_mq_create failed.\r\n", __FUNCTION__);return false;}return true;}rt_err_t rz_mq_magic_send(mq_magic_t *t, mq_magic_frame_t *buffer)
{if (RT_EOK != rt_mq_send(t->rt_mq, buffer, sizeof(mq_magic_frame_t))) {//rt_kprintf("%s(): rt_mq_send failed.\r\n", __FUNCTION__);return RT_ERROR;}return RT_EOK;
}rt_err_t rz_mq_magic_recv(mq_magic_t *t, void *buffer, int32_t timeout)
{if (RT_EOK != rt_mq_recv(t->rt_mq, buffer, sizeof(mq_magic_frame_t), timeout)) {//rt_kprintf("%s(): rz_mq_magic_recv failed.\r\n", __FUNCTION__);return RT_ERROR;}return RT_EOK;}/*timeout: uint is ms
*/
rt_err_t rz_mq_magic_send_wait_receive(pub_mq_magic_t *ppub, sub_mq_magic_t *psub, mq_magic_frame_t *buffer, int32_t timeout)
{
#define WAITTIME_BY_ONCE (100)//uint32_t timeout_num = 0;//uint32_t wait_count = 0;if (RT_EOK != rt_mq_send(ppub->rt_mq, buffer, sizeof(mq_magic_frame_t))) {rt_kprintf("%s(): rt_mq_send failed.\r\n", __FUNCTION__);return RT_ERROR;}return RT_EOK;#if 0timeout_num = timeout / WAITTIME_BY_ONCE;if (0 != (timeout % WAITTIME_BY_ONCE)) {timeout_num++;}memset(buffer, 0, sizeof(mq_magic_frame_t));do {wait_count++;if (RT_EOK == rt_mq_recv(psub->rt_mq, buffer, sizeof(mq_magic_frame_t), RT_WAITING_NO))return RT_EOK;rt_thread_mdelay(WAITTIME_BY_ONCE);} while (wait_count < timeout_num);return RT_ERROR;#endif
}/* init */
int rz_mq_magic_init(void)
{rz_mq_magic_t *prz_mq_magic = rz_mq_magic_obj_get();if (NULL != prz_mq_magic) {prz_mq_magic->init(prz_mq_magic);}return RT_EOK;}INIT_APP_EXPORT(rz_mq_magic_init);

这篇关于C语言实现设计模式—订阅者/发布者模式（Pub-Sub Pattern）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/188758。

相关文章

基于Python实现一个图片拆分工具

基于Python实现一个图片拆分工具

《基于Python实现一个图片拆分工具》这篇文章主要为大家详细介绍了如何基于Python实现一个图片拆分工具,可以根据需要的行数和列数进行拆分,感兴趣的小伙伴可以跟随小编一起学习一下... 简单介绍先自己选择输入的图片，默认是输出到项目文件夹中，可以自己选择其他的文件夹，选择需要拆分的行数和列数，可以通过
阅读更多...
Python中将嵌套列表扁平化的多种实现方法

Python中将嵌套列表扁平化的多种实现方法

《Python中将嵌套列表扁平化的多种实现方法》在Python编程中,我们常常会遇到需要将嵌套列表（即列表中包含列表）转换为一个一维的扁平列表的需求,本文将给大家介绍了多种实现这一目标的方法,需要的朋... 目录python中将嵌套列表扁平化的方法技术背景实现步骤1. 使用嵌套列表推导式2. 使用itert
阅读更多...
Python使用pip工具实现包自动更新的多种方法

Python使用pip工具实现包自动更新的多种方法

《Python使用pip工具实现包自动更新的多种方法》本文深入探讨了使用Python的pip工具实现包自动更新的各种方法和技术,我们将从基础概念开始,逐步介绍手动更新方法、自动化脚本编写、结合CI/C... 目录1. 背景介绍1.1 目的和范围1.2 预期读者1.3 文档结构概述1.4 术语表1.4.1 核
阅读更多...
在Linux中改变echo输出颜色的实现方法

在Linux中改变echo输出颜色的实现方法

《在Linux中改变echo输出颜色的实现方法》在Linux系统的命令行环境下,为了使输出信息更加清晰、突出,便于用户快速识别和区分不同类型的信息,常常需要改变echo命令的输出颜色,所以本文给大家介... 目python录在linux中改变echo输出颜色的方法技术背景实现步骤使用ANSI转义码使用tpu
阅读更多...
Python使用python-can实现合并BLF文件

Python使用python-can实现合并BLF文件

《Python使用python-can实现合并BLF文件》python-can库是Python生态中专注于CAN总线通信与数据处理的强大工具,本文将使用python-can为BLF文件合并提供高效灵活... 目录一、python-can 库：CAN 数据处理的利器二、BLF 文件合并核心代码解析1. 基础合
阅读更多...
Python使用OpenCV实现获取视频时长的小工具

Python使用OpenCV实现获取视频时长的小工具

《Python使用OpenCV实现获取视频时长的小工具》在处理视频数据时,获取视频的时长是一项常见且基础的需求,本文将详细介绍如何使用Python和OpenCV获取视频时长,并对每一行代码进行深入解析... 目录一、代码实现二、代码解析1. 导入 OpenCV 库2. 定义获取视频时长的函数3. 打开视频文
阅读更多...
golang版本升级如何实现

golang版本升级如何实现

《golang版本升级如何实现》：本文主要介绍golang版本升级如何实现问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录golanwww.chinasem.cng版本升级linux上golang版本升级删除golang旧版本安装golang最新版本总结gola
阅读更多...
Go语言中make和new的区别及说明

Go语言中make和new的区别及说明

《Go语言中make和new的区别及说明》：本文主要介绍Go语言中make和new的区别及说明,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1 概述2 new 函数2.1 功能2.2 语法2.3 初始化案例3 make 函数3.1 功能3.2 语法3.3 初始化
阅读更多...
SpringBoot中SM2公钥加密、私钥解密的实现示例详解

SpringBoot中SM2公钥加密、私钥解密的实现示例详解

《SpringBoot中SM2公钥加密、私钥解密的实现示例详解》本文介绍了如何在SpringBoot项目中实现SM2公钥加密和私钥解密的功能,通过使用Hutool库和BouncyCastle依赖,简化... 目录一、前言1、加密信息(示例)2、加密结果(示例)二、实现代码1、yml文件配置2、创建SM2工具
阅读更多...
Mysql实现范围分区表(新增、删除、重组、查看)

Mysql实现范围分区表(新增、删除、重组、查看)

《Mysql实现范围分区表(新增、删除、重组、查看)》MySQL分区表的四种类型（范围、哈希、列表、键值）,主要介绍了范围分区的创建、查询、添加、删除及重组织操作,具有一定的参考价值,感兴趣的可以了解... 目录一、mysql分区表分类二、范围分区（Range Partitioning1、新建分区表：2、分
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2