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使用watchdog实现文件资源监控

python使用watchdog实现文件资源监控

《python使用watchdog实现文件资源监控》watchdog支持跨平台文件资源监控,可以检测指定文件夹下文件及文件夹变动,下面我们来看看Python如何使用watchdog实现文件资源监控吧... python文件监控库watchdogs简介随着Python在各种应用领域中的广泛使用，其生态环境也
阅读更多...
el-select下拉选择缓存的实现

el-select下拉选择缓存的实现

《el-select下拉选择缓存的实现》本文主要介绍了在使用el-select实现下拉选择缓存时遇到的问题及解决方案,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的... 目录项目场景：问题描述解决方案：项目场景：从左侧列表中选取字段填入右侧下拉多选框，用户可以对右侧
阅读更多...
Python pyinstaller实现图形化打包工具

Python pyinstaller实现图形化打包工具

《Pythonpyinstaller实现图形化打包工具》：本文主要介绍一个使用PythonPYQT5制作的关于pyinstaller打包工具,代替传统的cmd黑窗口模式打包页面,实现更快捷方便的... 目录1.简介2.运行效果3.相关源码1.简介一个使用python PYQT5制作的关于pyinstall
阅读更多...
使用Python实现大文件切片上传及断点续传的方法

使用Python实现大文件切片上传及断点续传的方法

《使用Python实现大文件切片上传及断点续传的方法》本文介绍了使用Python实现大文件切片上传及断点续传的方法,包括功能模块划分（获取上传文件接口状态、临时文件夹状态信息、切片上传、切片合并）、整... 目录概要整体架构流程技术细节获取上传文件状态接口获取临时文件夹状态信息接口切片上传功能文件合并功能小
阅读更多...
python实现自动登录12306自动抢票功能

python实现自动登录12306自动抢票功能

《python实现自动登录12306自动抢票功能》随着互联网技术的发展,越来越多的人选择通过网络平台购票,特别是在中国,12306作为官方火车票预订平台,承担了巨大的访问量,对于热门线路或者节假日出行... 目录一、遇到的问题？二、改进三、进阶–展望总结一、遇到的问题？1.url-正确的表头：就是首先ur
阅读更多...
C#实现文件读写到SQLite数据库

C#实现文件读写到SQLite数据库

《C#实现文件读写到SQLite数据库》这篇文章主要为大家详细介绍了使用C#将文件读写到SQLite数据库的几种方法,文中的示例代码讲解详细,感兴趣的小伙伴可以参考一下... 目录1. 使用 BLOB 存储文件2. 存储文件路径3. 分块存储文件《文件读写到SQLite数据库China编程的方法》博客中，介绍了文
阅读更多...
Redis主从复制实现原理分析

Redis主从复制实现原理分析

《Redis主从复制实现原理分析》Redis主从复制通过Sync和CommandPropagate阶段实现数据同步,2.8版本后引入Psync指令,根据复制偏移量进行全量或部分同步,优化了数据传输效率... 目录Redis主DodMIK从复制实现原理实现原理Psync: 2.8版本后总结Redis主从复制实
阅读更多...
JAVA利用顺序表实现“杨辉三角”的思路及代码示例

JAVA利用顺序表实现“杨辉三角”的思路及代码示例

《JAVA利用顺序表实现“杨辉三角”的思路及代码示例》杨辉三角形是中国古代数学的杰出研究成果之一,是我国北宋数学家贾宪于1050年首先发现并使用的,：本文主要介绍JAVA利用顺序表实现杨辉三角的思... 目录一：“杨辉三角”题目链接二：题解代码：三：题解思路：总结一：“杨辉三角”题目链接题目链接：点击这里
阅读更多...
基于Python实现PDF动画翻页效果的阅读器

基于Python实现PDF动画翻页效果的阅读器

《基于Python实现PDF动画翻页效果的阅读器》在这篇博客中,我们将深入分析一个基于wxPython实现的PDF阅读器程序,该程序支持加载PDF文件并显示页面内容,同时支持页面切换动画效果,文中有详... 目录全部代码代码结构初始化 UI 界面加载 PDF 文件显示 PDF 页面页面切换动画运行效果总结主
阅读更多...
SpringBoot实现基于URL和IP的访问频率限制

SpringBoot实现基于URL和IP的访问频率限制

《SpringBoot实现基于URL和IP的访问频率限制》在现代Web应用中,接口被恶意刷新或暴力请求是一种常见的攻击手段,为了保护系统资源,需要对接口的访问频率进行限制,下面我们就来看看如何使用... 目录1. 引言2. 项目依赖3. 配置 Redis4. 创建拦截器5. 注册拦截器6. 创建控制器8.
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2