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进行终端交互式编程时,我们经常需要清空当前终端屏幕的内容,本文为大家整理了几种常见的实现方法,有需要的小伙伴可以参考下... 目录方法一：使用 `os` 模块调用系统命令方法二：使用 `subprocess` 模块执行命令方法三：打印多个换行符模拟
阅读更多...
SpringBoot+EasyPOI轻松实现Excel和Word导出PDF

SpringBoot+EasyPOI轻松实现Excel和Word导出PDF

《SpringBoot+EasyPOI轻松实现Excel和Word导出PDF》在企业级开发中,将Excel和Word文档导出为PDF是常见需求,本文将结合​​EasyPOI和​​Aspose系列工具实... 目录一、环境准备与依赖配置1.1 方案选型1.2 依赖配置（商业库方案）二、Excel 导出 PDF
阅读更多...
Python实现MQTT通信的示例代码

Python实现MQTT通信的示例代码

《Python实现MQTT通信的示例代码》本文主要介绍了Python实现MQTT通信的示例代码,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录1. 安装paho-mqtt库‌2. 搭建MQTT代理服务器（Broker）‌‌3. pytho
阅读更多...
使用zip4j实现Java中的ZIP文件加密压缩的操作方法

使用zip4j实现Java中的ZIP文件加密压缩的操作方法

《使用zip4j实现Java中的ZIP文件加密压缩的操作方法》本文介绍如何通过Maven集成zip4j1.3.2库创建带密码保护的ZIP文件,涵盖依赖配置、代码示例及加密原理,确保数据安全性,感兴趣的... 目录1. zip4j库介绍和版本1.1 zip4j库概述1.2 zip4j的版本演变1.3 zip4
阅读更多...
python生成随机唯一id的几种实现方法

python生成随机唯一id的几种实现方法

《python生成随机唯一id的几种实现方法》在Python中生成随机唯一ID有多种方法,根据不同的需求场景可以选择最适合的方案,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来一起学习学习... 目录方法 1：使用 UUID 模块（推荐）方法 2：使用 Secrets 模块（安全敏感场景）方法
阅读更多...
Spring StateMachine实现状态机使用示例详解

Spring StateMachine实现状态机使用示例详解

《SpringStateMachine实现状态机使用示例详解》本文介绍SpringStateMachine实现状态机的步骤,包括依赖导入、枚举定义、状态转移规则配置、上下文管理及服务调用示例,重点解... 目录什么是状态机使用示例什么是状态机状态机是计算机科学中的​​核心建模工具​​，用于描述对象在其生命
阅读更多...
Spring Boot 结合 WxJava 实现文章上传微信公众号草稿箱与群发

Spring Boot 结合 WxJava 实现文章上传微信公众号草稿箱与群发

《SpringBoot结合WxJava实现文章上传微信公众号草稿箱与群发》本文将详细介绍如何使用SpringBoot框架结合WxJava开发工具包,实现文章上传到微信公众号草稿箱以及群发功能,... 目录一、项目环境准备1.1 开发环境1.2 微信公众号准备二、Spring Boot 项目搭建2.1 创建
阅读更多...
IntelliJ IDEA2025创建SpringBoot项目的实现步骤

IntelliJ IDEA2025创建SpringBoot项目的实现步骤

《IntelliJIDEA2025创建SpringBoot项目的实现步骤》本文主要介绍了IntelliJIDEA2025创建SpringBoot项目的实现步骤,文中通过示例代码介绍的非常详细,对大家... 目录一、创建 Spring Boot 项目1. 新建项目2. 基础配置3. 选择依赖4. 生成项目5.
阅读更多...
深入理解Go语言中二维切片的使用

深入理解Go语言中二维切片的使用

《深入理解Go语言中二维切片的使用》本文深入讲解了Go语言中二维切片的概念与应用,用于表示矩阵、表格等二维数据结构,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来一起学习学习吧... 目录引言二维切片的基本概念定义创建二维切片二维切片的操作访问元素修改元素遍历二维切片二维切片的动态调整追加行动态
阅读更多...
Linux下删除乱码文件和目录的实现方式

Linux下删除乱码文件和目录的实现方式

《Linux下删除乱码文件和目录的实现方式》：本文主要介绍Linux下删除乱码文件和目录的实现方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录linux下删除乱码文件和目录方法1方法2总结Linux下删除乱码文件和目录方法1使用ls -i命令找到文件或目录
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2