在internet上建设网站可选择,浙江省职业建设学院官方网站,蕴川路上海网站建设,上海企业管理培训公司文章目录 0. 引言1. 设计概要1.1 主要组件1.2 类关系图1.3 工作流程 2. 代码实现2.1. 定义数据结构2.2. 实现 DataVisitor2.3. 实现 DataDispatcher2.4. 实现 Receiver2.5. 实现具体的 DataVisitor2.6. 示例主程序2.7. 编译和运行 0. 引言
使用 C 实现一个类似CyberRT 架构的 … 文章目录 0. 引言1. 设计概要1.1 主要组件1.2 类关系图1.3 工作流程 2. 代码实现2.1. 定义数据结构2.2. 实现 DataVisitor2.3. 实现 DataDispatcher2.4. 实现 Receiver2.5. 实现具体的 DataVisitor2.6. 示例主程序2.7. 编译和运行 0. 引言
使用 C 实现一个类似CyberRT 架构的 DataVisitor 和 DataDispatcher。在 CyberRT 中
Receiver 接收到消息后会触发回调。回调中调用 DataDispatcher消息分发器发布消息。DataDispatcher 是一个单例负责所有的数据分发并将数据放入对应的缓存中。然后DataDispatcher 会通知对应的协程在此简化为线程去处理消息。DataVisitor消息访问器是辅助类用于管理数据处理过程包括注册通知机制和绑定回调函数。
1. 设计概要
1.1 主要组件 DataDispatcher 单例模式。管理所有 DataVisitor。分发数据到对应的 DataVisitor 的缓冲区。通知 DataVisitor 处理数据。 DataVisitor 负责特定类型数据的处理。包含一个线程等待 DataDispatcher 的通知。绑定一个回调函数用于处理数据。管理自己的数据缓冲区。 Receiver 模拟消息接收器接收到消息后调用 DataDispatcher 发布数据。
1.2 类关系图
以下类关系图反映了 DataDispatcher 作为单例管理多个 DataVisitor并与 Receiver 交互的关系。 #mermaid-svg-u6RRQ3gFd6ATE5M6 {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .error-icon{fill:#552222;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .edge-thickness-normal{stroke-width:2px;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .marker.cross{stroke:#333333;}#mermaid-svg-u6RRQ3gFd6ATE5M6 svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-u6RRQ3gFd6ATE5M6 g.classGroup text{fill:#9370DB;fill:#131300;stroke:none;font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:10px;}#mermaid-svg-u6RRQ3gFd6ATE5M6 g.classGroup text .title{font-weight:bolder;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .nodeLabel,#mermaid-svg-u6RRQ3gFd6ATE5M6 .edgeLabel{color:#131300;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .edgeLabel .label rect{fill:#ECECFF;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .label text{fill:#131300;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .edgeLabel .label span{background:#ECECFF;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .classTitle{font-weight:bolder;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .node rect,#mermaid-svg-u6RRQ3gFd6ATE5M6 .node circle,#mermaid-svg-u6RRQ3gFd6ATE5M6 .node ellipse,#mermaid-svg-u6RRQ3gFd6ATE5M6 .node polygon,#mermaid-svg-u6RRQ3gFd6ATE5M6 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .divider{stroke:#9370DB;stroke:1;}#mermaid-svg-u6RRQ3gFd6ATE5M6 g.clickable{cursor:pointer;}#mermaid-svg-u6RRQ3gFd6ATE5M6 g.classGroup rect{fill:#ECECFF;stroke:#9370DB;}#mermaid-svg-u6RRQ3gFd6ATE5M6 g.classGroup line{stroke:#9370DB;stroke-width:1;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .classLabel .box{stroke:none;stroke-width:0;fill:#ECECFF;opacity:0.5;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .classLabel .label{fill:#9370DB;font-size:10px;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .relation{stroke:#333333;stroke-width:1;fill:none;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .dashed-line{stroke-dasharray:3;}#mermaid-svg-u6RRQ3gFd6ATE5M6 #compositionStart,#mermaid-svg-u6RRQ3gFd6ATE5M6 .composition{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-u6RRQ3gFd6ATE5M6 #compositionEnd,#mermaid-svg-u6RRQ3gFd6ATE5M6 .composition{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-u6RRQ3gFd6ATE5M6 #dependencyStart,#mermaid-svg-u6RRQ3gFd6ATE5M6 .dependency{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-u6RRQ3gFd6ATE5M6 #dependencyStart,#mermaid-svg-u6RRQ3gFd6ATE5M6 .dependency{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-u6RRQ3gFd6ATE5M6 #extensionStart,#mermaid-svg-u6RRQ3gFd6ATE5M6 .extension{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-u6RRQ3gFd6ATE5M6 #extensionEnd,#mermaid-svg-u6RRQ3gFd6ATE5M6 .extension{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-u6RRQ3gFd6ATE5M6 #aggregationStart,#mermaid-svg-u6RRQ3gFd6ATE5M6 .aggregation{fill:#ECECFF!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-u6RRQ3gFd6ATE5M6 #aggregationEnd,#mermaid-svg-u6RRQ3gFd6ATE5M6 .aggregation{fill:#ECECFF!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-u6RRQ3gFd6ATE5M6 .edgeTerminals{font-size:11px;}#mermaid-svg-u6RRQ3gFd6ATE5M6 :root{--mermaid-font-family:"trebuchet ms",verdana,arial,sans-serif;} 管理 1 * 调用 Dispatch Data int id std::string content «abstract» DataVisitor Notify(std::shared_ptrdata data) LoggingVisitor Create() ProcessingVisitor Create() DataDispatcher -std::vector visitors -std::mutex mutex Instance() RegisterVisitor(std::shared_ptr visitor) UnregisterVisitor(std::shared_ptr visitor) Dispatch(std::shared_ptrdata data) Receiver ReceiveMessage(int id, const std::string content) 1.3 工作流程
Receiver 接收到消息后调用 DataDispatcher::Instance()-Dispatch(data)。DataDispatcher 将数据放入对应的 DataVisitor 的缓冲区。DataDispatcher 通知对应的 DataVisitor。DataVisitor 的线程被唤醒取出数据并执行绑定的回调函数进行处理。
2. 代码实现
完整代码见data-visitor-dispatcher
2.1. 定义数据结构
首先定义一个通用的数据结构 Data。
// data.h
#ifndef DATA_H
#define DATA_H#include string// 定义一个通用的数据类型
struct Data {int id;std::string content;
};#endif // DATA_H
2.2. 实现 DataVisitor
DataVisitor 负责处理特定类型的数据。它包含一个线程该线程等待 DataDispatcher 的通知然后处理数据。
// data_visitor.h
#ifndef DATA_VISITOR_H
#define DATA_VISITOR_H#include atomic
#include condition_variable
#include functional
#include memory
#include mutex
#include queue
#include thread#include data.hclass DataVisitor : public std::enable_shared_from_thisDataVisitor {public:using Callback std::functionvoid(std::shared_ptrData);// 构造函数绑定回调函数explicit DataVisitor(Callback callback) : callback_(callback), stop_flag_(false) {worker_thread_ std::thread(DataVisitor::ProcessData, this);}// 析构函数确保线程安全停止~DataVisitor() {Stop();if (worker_thread_.joinable()) {worker_thread_.join();}}// 用于 DataDispatcher 发布数据时调用void Notify(std::shared_ptrData data) {{std::lock_guardstd::mutex lock(queue_mutex_);data_queue_.push(data);}cv_.notify_one();}private:// 数据处理线程函数void ProcessData() {while (!stop_flag_) {std::unique_lockstd::mutex lock(queue_mutex_);cv_.wait(lock, [this]() { return stop_flag_ || !data_queue_.empty(); });while (!data_queue_.empty()) {auto data data_queue_.front();data_queue_.pop();lock.unlock();// 执行绑定的回调函数if (callback_) {try {callback_(data);} catch (const std::exception e) {// 处理回调中的异常防止线程终止// 可以记录日志或采取其他措施// 这里简单输出错误信息std::cerr Exception in callback: e.what() std::endl;}}lock.lock();}}}// 停止线程void Stop() {stop_flag_ true;cv_.notify_all();}Callback callback_;std::thread worker_thread_;std::mutex queue_mutex_;std::condition_variable cv_;std::queuestd::shared_ptrData data_queue_;std::atomicbool stop_flag_;
};#endif // DATA_VISITOR_H
2.3. 实现 DataDispatcher
DataDispatcher 是一个单例负责管理所有的 DataVisitor 并分发数据。
// data_dispatcher.h
#ifndef DATA_DISPATCHER_H
#define DATA_DISPATCHER_H#include algorithm
#include memory
#include mutex
#include vector#include data.h
#include data_visitor.h// 单例的 DataDispatcher
class DataDispatcher {public:// 获取单例实例static DataDispatcher Instance() {static DataDispatcher instance;return instance;}// 禁止拷贝和赋值DataDispatcher(const DataDispatcher) delete;DataDispatcher operator(const DataDispatcher) delete;// 注册一个 DataVisitorvoid RegisterVisitor(const std::shared_ptrDataVisitor visitor) {std::lock_guardstd::mutex lock(mutex_);visitors_.emplace_back(visitor);}// 注销一个 DataVisitorvoid UnregisterVisitor(const std::shared_ptrDataVisitor visitor) {std::lock_guardstd::mutex lock(mutex_);visitors_.erase(std::remove(visitors_.begin(), visitors_.end(), visitor), visitors_.end());}// 分发数据到所有注册的 DataVisitorvoid Dispatch(const std::shared_ptrData data) {std::lock_guardstd::mutex lock(mutex_);for (auto visitor : visitors_) {if (visitor) {visitor-Notify(data);}}}private:// 私有构造函数DataDispatcher() default;~DataDispatcher() default;std::vectorstd::shared_ptrDataVisitor visitors_;std::mutex mutex_;
};#endif // DATA_DISPATCHER_H2.4. 实现 Receiver
模拟消息接收器每当接收到消息时调用 DataDispatcher 分发数据。
// receiver.h
#ifndef RECEIVER_H
#define RECEIVER_H#include functional
#include memory
#include string#include data.h
#include data_dispatcher.h// Receiver模拟消息接收器
class Receiver {public:using Callback std::functionvoid(std::shared_ptrData);// 构造函数绑定回调函数explicit Receiver(Callback callback) : callback_(callback) {}// 模拟接收消息void ReceiveMessage(int id, const std::string content) {auto data std::make_sharedData();data-id id;data-content content;// 触发回调if (callback_) {callback_(data);}}private:Callback callback_;
};#endif // RECEIVER_H2.5. 实现具体的 DataVisitor
例如创建 LoggingVisitor 和 ProcessingVisitor它们各自有不同的处理逻辑。
// logging_visitor.h
#ifndef LOGGING_VISITOR_H
#define LOGGING_VISITOR_H#include iostream
#include memory
#include data_visitor.h// LoggingVisitor负责记录数据
class LoggingVisitor {public:// 创建一个 LoggingVisitor 的 DataVisitor 实例static std::shared_ptrDataVisitor Create() {return std::make_sharedDataVisitor([](std::shared_ptrData data) {std::cout [LoggingVisitor] Received data: ID data-id , Content\ data-content \ std::endl;});}
};#endif // LOGGING_VISITOR_H// processing_visitor.h
#ifndef PROCESSING_VISITOR_H
#define PROCESSING_VISITOR_H#include iostream
#include memory
#include data_visitor.h// ProcessingVisitor负责处理数据
class ProcessingVisitor {public:// 创建一个 ProcessingVisitor 的 DataVisitor 实例static std::shared_ptrDataVisitor Create() {return std::make_sharedDataVisitor([](std::shared_ptrData data) {// 简单示例打印数据长度std::cout [ProcessingVisitor] Processed data ID data-id , Length data-content.length() std::endl;});}
};#endif // PROCESSING_VISITOR_H2.6. 示例主程序
展示如何使用上述组件实现数据接收、分发和处理。
// main.cpp
#include chrono
#include iostream
#include memory
#include thread#include data_dispatcher.h
#include logging_visitor.h
#include processing_visitor.h
#include receiver.hint main() {// 创建并注册 DataVisitorauto logger LoggingVisitor::Create();auto processor ProcessingVisitor::Create();DataDispatcher::Instance().RegisterVisitor(logger);DataDispatcher::Instance().RegisterVisitor(processor);// 创建 Receiver并绑定 DataDispatcher 的 Dispatch 方法Receiver receiver([](std::shared_ptrData data) { DataDispatcher::Instance().Dispatch(data); });// 模拟接收消息std::cout 接收第1条消息 std::endl;receiver.ReceiveMessage(1, Hello, CyberRT!);std::cout 接收第2条消息 std::endl;receiver.ReceiveMessage(2, Another data packet.);// 等待一段时间以确保所有消息被处理std::this_thread::sleep_for(std::chrono::seconds(1));// 注销一个 DataVisitorstd::cout \n 移除 LoggingVisitor std::endl;DataDispatcher::Instance().UnregisterVisitor(logger);// 再次接收消息std::cout 接收第3条消息 std::endl;receiver.ReceiveMessage(3, Data after removing logger.);// 等待一段时间以确保消息被处理std::this_thread::sleep_for(std::chrono::seconds(1));std::cout \n 程序结束 std::endl;return 0;
}2.7. 编译和运行
假设所有文件在同一目录下可以使用以下命令进行编译
g -stdc14 main.cpp -o dispatcher -pthread运行程序后您将看到类似如下的输出
testpi:~/dataVisitor$ g -stdc14 main.cpp -o dispatcher -pthread
testpi:~/dataVisitor$ ./dispatcher 接收第1条消息 接收第2条消息
[LoggingVisitor] Received data: ID[ProcessingVisitor] Processed data ID11, ContentHello, CyberRT!
[LoggingVisitor] Received data: ID2, ContentAnother data packet.
, Length15
[ProcessingVisitor] Processed data ID2, Length20 移除 LoggingVisitor 接收第3条消息
[ProcessingVisitor] Processed data ID3, Length27 程序结束 注意第三条消息只被 ProcessingVisitor 处理因为 LoggingVisitor 已被移除。
