架构师之路:用C++实现一个消息路由系统
liuian 2024-12-07 14:59 52 浏览
引言
设计和实现高效的消息路由系统是一名架构师的基本功和重要任务之一。本文将深入浅出探讨基于C++的消息路由的设计原则和实现方法。逐步介绍消息路由的概念、需求分析、设计思路和关键实现细节，并通过具体的示例来说明。
第一部分：消息路由的概念和需求分析
1.1 消息路由的定义
消息路由是将消息从发送者传递到接收者的过程，通常用于多个模块或进程之间的通信。在的系统中，消息可以是任何可序列化的数据结构，如JSON、Protocol Buffers等。
1.2需求分析
在设计消息路由系统时，明确需求非常关键，以下是对需求的进一步分析：
- 可靠性：为了保证可靠性，可以引入消息确认机制，发送者在发送消息后等待接收者的确认，以确保消息已被成功处理。如果没有确认，发送者可以选择重新发送消息。
- 灵活性：实现灵活的消息处理，可以考虑引入消息过滤器，订阅者可以根据消息内容、类型或其他标志来选择接收哪些消息。
- 扩展性：确保系统具有良好的扩展性，可以将订阅关系和消息队列分布在不同的节点上，以支持大规模和高并发的消息传递。
第二部分：设计思路
2.1 消息路由的架构
在设计消息路由系统的架构时，需要考虑以下几个方面：
- 消息路由器：消息路由器是系统的核心组件，负责接收和分发消息。它需要能够处理大规模和高并发的消息传递，并保证消息的可靠性。消息路由器可以采用单节点或分布式部署，以满足系统的性能和可扩展性需求。
- 订阅关系：订阅关系是消息路由器用来确定消息应该传递给哪些订阅者的重要依据。订阅关系可以由用户自行管理，也可以通过注册中心来管理。在订阅关系中，可以定义不同类型的订阅，如精确订阅、模糊订阅等，以满足用户的个性化需求。
- 消息传递协议：消息传递协议定义了消息的格式和传递方式。常见的消息传递协议有HTTP、TCP、AMQP等。选择合适的消息传递协议可以提高系统的性能和可靠性。
在下面的C++示例中，我将演示如何结合消息路由器、订阅关系和UDP消息传递协议来设计一个基本的消息路由系统。
#include <iostream>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <unordered_map>
#include <vector>
// 消息结构
struct Message {
std::string topic;
std::string content;
};
// 订阅者类
class Subscriber {
public:
Subscriber(const std::string& name) : name(name) {}
void receiveMessage(const Message& message) {
std::cout << name << " received message on topic '" << message.topic << "': " << message.content << std::endl;
}
private:
std::string name;
};
// 消息路由器类
class MessageRouter {
public:
void subscribe(const Subscriber& subscriber, const std::string& topic) {
subscribers[topic].push_back(subscriber);
}
void publish(const Message& message) {
const std::string& topic = message.topic;
if (subscribers.find(topic) != subscribers.end()) {
for (const auto& subscriber : subscribers[topic]) {
subscriber.receiveMessage(message);
}
}
}
private:
std::unordered_map<std::string, std::vector<Subscriber>> subscribers;
};
int main() {
// 创建消息路由器
MessageRouter router;
// 创建订阅者
Subscriber subscriber1("Subscriber 1");
Subscriber subscriber2("Subscriber 2");
Subscriber subscriber3("Subscriber 3");
// 订阅关系
router.subscribe(subscriber1, "news");
router.subscribe(subscriber2, "sports");
router.subscribe(subscriber3, "news");
// 创建UDP套接字
int udpSocket = socket(AF_INET, SOCK_DGRAM, 0);
if (udpSocket < 0) {
std::cerr << "Error creating UDP socket" << std::endl;
return 1;
}
// 设置服务器地址和端口
struct sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_addr.s_addr = INADDR_ANY;
serverAddress.sin_port = htons(9000); // 选择一个合适的端口号
// 绑定套接字到服务器地址和端口
if (bind(udpSocket, (struct sockaddr*)&serverAddress, sizeof(serverAddress)) < 0) {
std::cerr << "Error binding UDP socket" << std::endl;
return 1;
}
std::cout << "UDP server is running on port 9000" << std::endl;
// 接收和处理消息
char buffer[1024];
struct sockaddr_in clientAddress;
socklen_t clientAddressLength = sizeof(clientAddress);
while (true) {
memset(buffer, 0, sizeof(buffer));
// 接收消息
ssize_t bytesRead = recvfrom(udpSocket, buffer, sizeof(buffer), 0, (struct sockaddr*)&clientAddress, &clientAddressLength);
if (bytesRead < 0) {
std::cerr << "Error receiving message" << std::endl;
continue;
}
// 处理消息
std::string messageContent(buffer, bytesRead);
std::string messageTopic = "unknown"; // 默认主题
// 解析消息内容，假设消息格式为 "Topic: Content"
size_t separatorPos = messageContent.find(":");
if (separatorPos != std::string::npos) {
messageTopic = messageContent.substr(0, separatorPos);
messageContent = messageContent.substr(separatorPos + 1);
}
Message message = {messageTopic, messageContent};
router.publish(message); // 将消息路由到订阅者
std::cout << "Received UDP message: " << messageContent << " from " << inet_ntoa(clientAddress.sin_addr) << ":" << ntohs(clientAddress.sin_port) << std::endl;
}
// 关闭套接字
close(udpSocket);
return 0;
}上述代码首先创建了一个消息路由器 MessageRouter，并创建了三个订阅者。然后，使用UDP协议创建了一个UDP套接字，并绑定到指定的端口（这里使用9000端口）来接收消息。当接收到UDP消息时，解析消息内容，将消息路由到相应的订阅者，同时也打印了接收到的UDP消息的内容和发送方的信息。这个示例中包含了消息路由器、订阅关系和UDP消息传递协议的集成。
2.2 订阅关系的管理
为了管理订阅关系，可以引入一个注册中心。注册中心可以使用分布式数据库或分布式键值存储来实现，以确保高可用性和可扩展性。注册中心需要提供API，让模块能够注册和查询订阅关系。
注册中心可以有以下功能：
- 注册订阅关系：订阅者可以向注册中心注册自己的订阅关系，包括订阅的主题、订阅者的标识等。注册中心将订阅关系存储起来，以便消息路由器查询和使用。
- 查询订阅关系：消息路由器可以向注册中心查询某个主题的订阅者列表，以便将消息传递给相应的订阅者。
- 更新订阅关系：订阅者可以随时更新自己的订阅关系，例如添加新的订阅、取消已有的订阅等。注册中心需要及时更新订阅关系，以保证消息的正确传递。
#include <iostream>
#include <string>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <unordered_map>
#include <vector>
// 消息结构
struct Message {
std::string topic;
std::string content;
};
// 订阅者类
class Subscriber {
public:
Subscriber(const std::string& name) : name(name) {}
void receiveMessage(const Message& message) {
std::cout << name << " received message on topic '" << message.topic << "': " << message.content << std::endl;
}
std::string getName() const {
return name;
}
private:
std::string name;
};
// 消息路由器类
class MessageRouter {
public:
void subscribe(const Subscriber& subscriber, const std::string& topic) {
subscriptions.registerSubscription({topic, subscriber.getName()});
}
void publish(const Message& message) {
const std::string& topic = message.topic;
std::vector<std::string> subscribers = subscriptions.getSubscribers(topic);
for (const auto& subscriber : subscribers) {
for (const auto& s : subscribersList) {
if (s.getName() == subscriber) {
s.receiveMessage(message);
}
}
}
}
private:
Registry subscriptions; // 使用注册中心管理订阅关系
std::vector<Subscriber> subscribersList; // 订阅者列表
};
// 订阅关系结构
struct Subscription {
std::string topic;
std::string subscriber;
};
// 注册中心类
class Registry {
public:
//注册订阅关系
void registerSubscription(const Subscription& subscription) {
subscriptions[subscription.topic].push_back(subscription.subscriber);
}
//查找订阅关系
std::vector<std::string> getSubscribers(const std::string& topic) {
if (subscriptions.find(topic) != subscriptions.end()) {
return subscriptions[topic];
}
return {};
}
//更新订阅关系
void updateSubscription(const Subscription& oldSubscription, const Subscription& newSubscription) {
auto& topicSubscribers = subscriptions[oldSubscription.topic];
for (auto it = topicSubscribers.begin(); it != topicSubscribers.end(); ++it) {
if (*it == oldSubscription.subscriber) {
*it = newSubscription.subscriber;
}
}
}
private:
std::unordered_map<std::string, std::vector<std::string>> subscriptions;
};
int main() {
// 创建消息路由器
MessageRouter router;
// 创建订阅者
Subscriber subscriber1("Subscriber 1");
Subscriber subscriber2("Subscriber 2");
Subscriber subscriber3("Subscriber 3");
// 订阅关系示例
Subscription subscription1 = {"news", "Subscriber 1"};
Subscription subscription2 = {"sports", "Subscriber 2"};
Subscription subscription3 = {"news", "Subscriber 3"};
// 注册订阅关系
router.subscribe(subscriber1, "news");
router.subscribe(subscriber2, "sports");
router.subscribe(subscriber3, "news");
// 创建UDP套接字
int udpSocket = socket(AF_INET, SOCK_DGRAM, 0);
if (udpSocket < 0) {
std::cerr << "Error creating UDP socket" << std::endl;
return 1;
}
// 设置服务器地址和端口
struct sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_addr.s_addr = INADDR_ANY;
serverAddress.sin_port = htons(9000); // 选择一个合适的端口号
// 绑定套接字到服务器地址和端口
if (bind(udpSocket, (struct sockaddr*)&serverAddress, sizeof(serverAddress)) < 0) {
std::cerr << "Error binding UDP socket" << std::endl;
return 1;
}
std::cout << "UDP server is running on port 9000" << std::endl;
// 接收和处理消息
char buffer[1024];
struct sockaddr_in clientAddress;
socklen_t clientAddressLength = sizeof(clientAddress);
while (true) {
memset(buffer, 0, sizeof(buffer));
// 接收消息
ssize_t bytesRead = recvfrom(udpSocket, buffer, sizeof(buffer), 0, (struct sockaddr*)&clientAddress, &clientAddressLength);
if (bytesRead < 0) {
std::cerr << "Error receiving message" << std::endl;
continue;
}
// 处理消息
std::string messageContent(buffer, bytesRead);
std::string messageTopic = "unknown"; // 默认主题
// 解析消息内容，假设消息格式为 "Topic: Content"
size_t separatorPos = messageContent.find(":");
if (separatorPos != std::string::npos) {
messageTopic = messageContent.substr(0, separatorPos);
messageContent = messageContent.substr(separatorPos + 1);
}
Message message = {messageTopic, messageContent};
router.publish(message); // 将消息路由到订阅者
std::cout << "Received UDP message: " << messageContent << " from " << inet_ntoa(clientAddress.sin_addr) << ":" << ntohs(clientAddress.sin_port) << std::endl;
}
// 关闭套接字
close(udpSocket);
return 0;
}上述代码将 Registry 类嵌入到 MessageRouter 类中，以管理订阅关系。当订阅者通过 subscribe 方法注册订阅关系时，会将订阅信息添加到注册中心。在 publish 方法中，消息路由器会从注册中心查询订阅者列表，并将消息传递给相应的订阅者。这样，订阅关系管理被整合到消息路由系统中。
2.3 消息的传递和处理
消息路由器接收到消息后，需要将消息传递给相应的订阅者。为了实现高性能和可靠性的消息传递，可以采用以下策略：
- 多线程处理：可以为每个订阅者分配一个独立的线程来处理消息。这样可以实现并发处理，提高系统的吞吐量。需要注意线程安全性，可以采用互斥锁或其他同步机制来保护共享资源。
- 消息队列服务：可以选择使用成熟的消息队列服务，如RabbitMQ、Kafka等。它们提供了高度可靠的消息传递机制，并且能够处理大规模和高并发的消息传递。消息路由器将消息发送到消息队列中，订阅者从消息队列中取出消息并进行处理。
- 消息持久化：为了保证消息的可靠传递，可以将消息持久化到存储介质中，如数据库、文件系统等。这样即使系统发生故障，消息也不会丢失。可以使用事务或其他机制来保证消息的一致性和可靠性。
#include <iostream>
#include <string>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <unordered_map>
#include <vector>
#include <thread>
#include <mutex>
#include <queue>
#include <condition_variable>
#include <sqlite3.h> // 添加SQLite库的头文件
// 消息结构
struct Message {
std::string topic;
std::string content;
};
// 订阅者类
class Subscriber {
public:
Subscriber(const std::string& name) : name(name) {}
void receiveMessage(const Message& message) {
std::cout << name << " received message on topic '" << message.topic << "': " << message.content << std::endl;
// 处理完消息后从数据库中删除消息
deleteMessageFromDatabase(message);
}
std::string getName() const {
return name;
}
private:
std::string name;
void deleteMessageFromDatabase(const Message& message) {
sqlite3* db;
if (sqlite3_open("messages.db", &db) == SQLITE_OK) {
std::string deleteQuery = "DELETE FROM messages WHERE topic = ? AND content = ?";
sqlite3_stmt* stmt;
if (sqlite3_prepare_v2(db, deleteQuery.c_str(), -1, &stmt, NULL) == SQLITE_OK) {
sqlite3_bind_text(stmt, 1, message.topic.c_str(), -1, SQLITE_STATIC);
sqlite3_bind_text(stmt, 2, message.content.c_str(), -1, SQLITE_STATIC);
sqlite3_step(stmt);
sqlite3_finalize(stmt);
}
sqlite3_close(db);
}
}
};
// 消息队列类
class MessageQueue {
public:
void push(const Message& message) {
std::lock_guard<std::mutex> lock(mutex);
queue.push(message);
condition.notify_one();
// 消息入队后存储到数据库
storeMessageToDatabase(message);
}
Message pop() {
std::unique_lock<std::mutex> lock(mutex);
condition.wait(lock, [this] { return !queue.empty(); });
Message message = queue.front();
queue.pop();
return message;
}
private:
std::queue<Message> queue;
std::mutex mutex;
std::condition_variable condition;
void storeMessageToDatabase(const Message& message) {
sqlite3* db;
if (sqlite3_open("messages.db", &db) == SQLITE_OK) {
std::string insertQuery = "INSERT INTO messages (topic, content) VALUES (?, ?)";
sqlite3_stmt* stmt;
if (sqlite3_prepare_v2(db, insertQuery.c_str(), -1, &stmt, NULL) == SQLITE_OK) {
sqlite3_bind_text(stmt, 1, message.topic.c_str(), -1, SQLITE_STATIC);
sqlite3_bind_text(stmt, 2, message.content.c_str(), -1, SQLITE_STATIC);
sqlite3_step(stmt);
sqlite3_finalize(stmt);
}
sqlite3_close(db);
}
}
};
// 消息路由器类
class MessageRouter {
public:
MessageRouter() : queue(new MessageQueue()) {
initializeDatabase();
}
void subscribe(const Subscriber& subscriber, const std::string& topic) {
subscriptions.registerSubscription({topic, subscriber.getName()});
createThreadForSubscriber(subscriber);
}
void publish(const Message& message) {
const std::string& topic = message.topic;
std::vector<std::string> subscribers = subscriptions.getSubscribers(topic);
for (const auto& subscriber : subscribers) {
queue->push(message); // 将消息推送到消息队列
}
}
private:
Registry subscriptions; // 使用注册中心管理订阅关系
std::unique_ptr<MessageQueue> queue; // 消息队列
std::vector<Subscriber> subscribersList; // 订阅者列表
std::vector<std::thread> subscriberThreads; // 订阅者线程列表
void createThreadForSubscriber(const Subscriber& subscriber) {
// 为每个订阅者创建一个独立的线程
subscriberThreads.push_back(std::thread([this, &subscriber] {
while (true) {
// 从消息队列中取出消息并处理
Message message = queue->pop();
subscriber.receiveMessage(message);
}
}));
}
void initializeDatabase() {
// 初始化SQLite数据库
sqlite3* db;
if (sqlite3_open("messages.db", &db) == SQLITE_OK) {
const char* createTableQuery = "CREATE TABLE IF NOT EXISTS messages (topic TEXT, content TEXT)";
sqlite3_exec(db, createTableQuery, 0, 0, 0);
sqlite3_close(db);
}
}
};
// 订阅关系结构
struct Subscription {
std::string topic;
std::string subscriber;
};
// 注册中心类
class Registry {
public:
void registerSubscription(const Subscription& subscription) {
subscriptions[subscription.topic].push_back(subscription.subscriber);
}
std::vector<std::string> getSubscribers(const std::string& topic) {
if (subscriptions.find(topic) != subscriptions.end()) {
return subscriptions[topic];
}
return {};
}
void updateSubscription(const Subscription& oldSubscription, const Subscription& newSubscription) {
auto& topicSubscribers = subscriptions[oldSubscription.topic];
for (auto it = topicSubscribers.begin(); it != topicSubscribers.end(); ++it) {
if (*it == oldSubscription.subscriber) {
*it = newSubscription.subscriber;
}
}
}
private:
std::unordered_map<std::string, std::vector<std::string>> subscriptions;
};
int main() {
// 创建消息路由器
MessageRouter router;
// 创建订阅者
Subscriber subscriber1("Subscriber 1");
Subscriber subscriber2("Subscriber 2");
Subscriber subscriber3("Subscriber 3");
// 订阅关系示例
Subscription subscription1 = {"news", "Subscriber 1"};
Subscription subscription2 = {"sports", "Subscriber 2"};
Subscription subscription3 = {"news", "Subscriber 3"};
// 注册订阅关系
router.subscribe(subscriber1, "news");
router.subscribe(subscriber2, "sports");
router.subscribe(subscriber3, "news");
// 创建UDP套接字
int udpSocket = socket(AF_INET, SOCK_DGRAM, 0);
if (udpSocket < 0) {
std::cerr << "Error creating UDP socket" << std::endl;
return 1;
}
// 设置服务器地址和端口
struct sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_addr.s_addr = INADDR_ANY;
serverAddress.sin_port = htons(9000); // 选择一个合适的端口号
// 绑定套接字到服务器地址和端口
if (bind(udpSocket, (struct sockaddr*)&serverAddress, sizeof(serverAddress)) < 0) {
std::cerr << "Error binding UDP socket" << std::endl;
return 1;
}
std::cout << "UDP server is running on port 9000" << std::endl;
// 接收和处理消息
char buffer[1024];
struct sockaddr_in clientAddress;
socklen_t clientAddressLength = sizeof(clientAddress);
while (true) {
memset(buffer, 0, sizeof(buffer));
// 接收消息
ssize_t bytesRead = recvfrom(udpSocket, buffer, sizeof(buffer), 0, (struct sockaddr*)&clientAddress, &clientAddressLength);
if (bytesRead < 0) {
std::cerr << "Error receiving message" << std::endl;
continue;
}
// 处理消息
std::string messageContent(buffer, bytesRead);
std::string messageTopic = "unknown"; // 默认主题
// 解析消息内容，假设消息格式为 "Topic: Content"
size_t separatorPos = messageContent.find(":");
if (separatorPos != std::string::npos) {
messageTopic = messageContent.substr(0, separatorPos);
messageContent = messageContent.substr(separatorPos + 1);
}
Message message = {messageTopic, messageContent};
router.publish(message); // 将消息路由到订阅者
std::cout << "Received UDP message: " << messageContent << " from " << inet_ntoa(clientAddress.sin_addr) << ":" << ntohs(clientAddress.sin_port) << std::endl;
}
// 关闭套接字
close(udpSocket);
return 0;
}
上述代码上述代码引入了一个 MessageQueue 类来模拟消息队列的功能。每个订阅者都有一个独立的线程，从消息队列中获取消息并进行处理。消息的发布操作将消息推送到消息队列中，由订阅者线程异步处理。这样可以提高消息传递的性能和可靠性。使用了SQLite数据库来存储消息。当订阅者接收到消息后，会从数据库中删除该消息。在初始化阶段，创建了一个名为 "messages.db" 的SQLite数据库，并创建了一个名为 "messages" 的表来存储消息。
第五部分：示例
以上是一个简单的消息传递和处理系统的设计和实现思路，具体的实现细节还需要根据具体的需求和场景进行调整和完善。下面将对示例中的电商系统可以进一步细化为以下步骤：
- 商品服务将商品信息发布到消息路由器。
- 订单服务和用户服务在启动时向注册中心注册对商品信息消息的订阅关系。
- 消息路由器接收到商品信息消息后，查询注册中心，将消息分发给订单服务和用户服务。
- 订单服务和用户服务从消息队列中取出商品信息消息并进行相应的处理，例如生成订单或者更新用户信息。
首先，从创建商品服务并发布商品信息消息开始。以下是商品服务的代码示例：
// 商品服务
#include <iostream>
#include <string>
#include <unistd.h>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
// 商品信息结构
struct ProductInfo {
std::string productId;
std::string name;
double price;
int stock;
};
// 函数用于发布商品信息消息
void publishProductInfo(const ProductInfo& productInfo) {
// 创建UDP套接字
int udpSocket = socket(AF_INET, SOCK_DGRAM, 0);
if (udpSocket < 0) {
std::cerr << "Error creating UDP socket" << std::endl;
return;
}
// 设置消息服务器地址和端口
struct sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_addr.s_addr = inet_addr("127.0.0.1"); // 假设消息服务器在本地运行
serverAddress.sin_port = htons(9000); // 选择一个合适的端口号
// 准备商品信息消息
std::string message = "ProductInfo:" + productInfo.productId + ":" + productInfo.name + ":" +
std::to_string(productInfo.price) + ":" + std::to_string(productInfo.stock);
// 发送消息
ssize_t bytesSent = sendto(udpSocket, message.c_str(), message.size(), 0, (struct sockaddr*)&serverAddress, sizeof(serverAddress));
if (bytesSent < 0) {
std::cerr << "Error sending message" << std::endl;
} else {
std::cout << "Published product information: " << productInfo.name << std::endl;
}
// 关闭套接字
close(udpSocket);
}
int main() {
// 模拟商品信息发布
ProductInfo product1 = {"123", "Product A", 19.99, 100};
ProductInfo product2 = {"456", "Product B", 29.99, 50};
// 发布商品信息消息
publishProductInfo(product1);
publishProductInfo(product2);
return 0;
}在上述代码中，商品服务模拟了两个商品的信息，并使用UDP套接字将商品信息消息发布到消息服务器（本地地址和端口9000）。每个商品信息消息的格式是 "ProductInfo:productId:name:price:stock"。商品服务将商品信息消息发送后，消息路由器会将它分发给订阅了商品信息的服务（例如订单服务和用户服务）。
接下来，将创建订单服务和用户服务，并在它们启动时注册对商品信息消息的订阅关系。以下是订单服务和用户服务的代码示例：
订单服务（Order Service）：
// 订单服务
#include <iostream>
#include <string>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <vector>
// 消息结构
struct Message {
std::string topic;
std::string content;
};
// 订阅者类
class Subscriber {
public:
Subscriber(const std::string& name) : name(name) {}
void receiveMessage(const Message& message) {
std::cout << name << " received message on topic '" << message.topic << "': " << message.content << std::endl;
// 在实际应用中，这里可以处理订单生成等业务逻辑
}
std::string getName() const {
return name;
}
private:
std::string name;
};
int main() {
// 创建订单服务
Subscriber orderService("Order Service");
// 创建UDP套接字
int udpSocket = socket(AF_INET, SOCK_DGRAM, 0);
if (udpSocket < 0) {
std::cerr << "Error creating UDP socket" << std::endl;
return 1;
}
// 设置服务器地址和端口
struct sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_addr.s_addr = INADDR_ANY;
serverAddress.sin_port = htons(9000); // 选择一个合适的端口号
// 绑定套接字到服务器地址和端口
if (bind(udpSocket, (struct sockaddr*)&serverAddress, sizeof(serverAddress)) < 0) {
std::cerr << "Error binding UDP socket" << std::endl;
return 1;
}
std::cout << "Order Service is running and subscribed to product information" << std::endl;
// 接收和处理消息
char buffer[1024];
struct sockaddr_in clientAddress;
socklen_t clientAddressLength = sizeof(clientAddress);
while (true) {
memset(buffer, 0, sizeof(buffer));
// 接收消息
ssize_t bytesRead = recvfrom(udpSocket, buffer, sizeof(buffer), 0, (struct sockaddr*)&clientAddress, &clientAddressLength);
if (bytesRead < 0) {
std::cerr << "Error receiving message" << std::endl;
continue;
}
// 处理消息
std::string messageContent(buffer, bytesRead);
std::string messageTopic = "unknown"; // 默认主题
// 解析消息内容，假设消息格式为 "Topic: Content"
size_t separatorPos = messageContent.find(":");
if (separatorPos != std::string::npos) {
messageTopic = messageContent.substr(0, separatorPos);
messageContent = messageContent.substr(separatorPos + 1);
}
Message message = {messageTopic, messageContent};
orderService.receiveMessage(message); // 处理消息
}
// 关闭套接字
close(udpSocket);
return 0;
}用户服务（User Service）：
// 用户服务
#include <iostream>
#include <string>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <vector>
// 消息结构
struct Message {
std::string topic;
std::string content;
};
// 订阅者类
class Subscriber {
public:
Subscriber(const std::string& name) : name(name) {}
void receiveMessage(const Message& message) {
std::cout << name << " received message on topic '" << message.topic << "': " << message.content << std::endl;
// 在实际应用中，这里可以处理用户信息更新等业务逻辑
}
std::string getName() const {
return name;
}
private:
std::string name;
};
int main() {
// 创建用户服务
Subscriber userService("User Service");
// 创建UDP套接字
int udpSocket = socket(AF_INET, SOCK_DGRAM, 0);
if (udpSocket < 0) {
std::cerr << "Error creating UDP socket" << std::endl;
return 1;
}
// 设置服务器地址和端口
struct sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_addr.s_addr = INADDR_ANY;
serverAddress.sin_port = htons(9000); // 选择一个合适的端口号
// 绑定套接字到服务器地址和端口
if (bind(udpSocket, (struct sockaddr*)&serverAddress, sizeof(serverAddress)) < 0) {
std::cerr << "Error binding UDP socket" << std::endl;
return 1;
}
std::cout << "User Service is running and subscribed to product information" << std::endl;
// 接收和处理消息
char buffer[1024];
struct sockaddr_in clientAddress;
socklen_t clientAddressLength = sizeof(clientAddress);
while (true) {
memset(buffer, 0, sizeof(buffer));
// 接收消息
ssize_t bytesRead = recvfrom(udpSocket, buffer, sizeof(buffer), 0, (struct sockaddr*)&clientAddress, &clientAddressLength);
if (bytesRead < 0) {
std::cerr << "Error receiving message" << std::endl;
continue;
}
// 处理消息
std::string messageContent(buffer, bytesRead);
std::string messageTopic = "unknown"; // 默认主题
// 解析消息内容，假设消息格式为 "Topic: Content"
size_t separatorPos = messageContent.find(":");
if (separatorPos != std::string::npos) {
messageTopic = messageContent.substr(0, separatorPos);
messageContent = messageContent.substr(separatorPos + 1);
}
Message message = {messageTopic, messageContent};
userService.receiveMessage(message); // 处理消息
}
// 关闭套接字
close(udpSocket);
return 0;
}在上述代码中，订单服务和用户服务都创建了一个UDP套接字，以接收商品信息消息。它们在启动时分别注册了对商品信息消息的订阅关系，并可以在 receiveMessage 方法中处理消息，执行订单生成或用户信息更新等业务逻辑。
接下来，将创建一个消息路由器（Message Router）来处理和分发消息。消息路由器将接收商品信息消息，并将其分发给已经注册了对该消息的订阅关系的订单服务和用户服务。
// 消息路由器
#include <iostream>
#include <string>
#include <cstring>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <vector>
#include <thread>
#include <mutex>
#include <queue>
#include <condition_variable>
// 消息结构
struct Message {
std::string topic;
std::string content;
};
// 订阅者类
class Subscriber {
public:
Subscriber(const std::string& name) : name(name) {}
void receiveMessage(const Message& message) {
std::cout << name << " received message on topic '" << message.topic << "': " << message.content << std::endl;
// 在实际应用中，这里可以处理订单生成或用户信息更新等业务逻辑
}
std::string getName() const {
return name;
}
private:
std::string name;
};
// 消息队列类
class MessageQueue {
public:
void push(const Message& message) {
std::lock_guard<std::mutex> lock(mutex);
queue.push(message);
condition.notify_one();
}
Message pop() {
std::unique_lock<std::mutex> lock(mutex);
condition.wait(lock, [this] { return !queue.empty(); });
Message message = queue.front();
queue.pop();
return message;
}
private:
std::queue<Message> queue;
std::mutex mutex;
std::condition_variable condition;
};
// 消息路由器类
class MessageRouter {
public:
MessageRouter() : queue(new MessageQueue()) {}
void subscribe(const Subscriber& subscriber, const std::string& topic) {
subscriptions.registerSubscription({topic, subscriber.getName()});
createThreadForSubscriber(subscriber);
}
void publish(const Message& message) {
const std::string& topic = message.topic;
std::vector<std::string> subscribers = subscriptions.getSubscribers(topic);
for (const auto& subscriber : subscribers) {
queue->push(message); // 将消息推送到消息队列
}
}
void start() {
// 启动消息处理线程
messageProcessingThread = std::thread([this] {
while (true) {
// 从消息队列中取出消息并处理
Message message = queue->pop();
dispatchMessage(message);
}
});
}
void stop() {
// 停止消息处理线程
if (messageProcessingThread.joinable()) {
messageProcessingThread.join();
}
}
private:
Registry subscriptions; // 使用注册中心管理订阅关系
std::unique_ptr<MessageQueue> queue; // 消息队列
std::thread messageProcessingThread; // 消息处理线程
void createThreadForSubscriber(const Subscriber& subscriber) {
// 为每个订阅者创建一个独立的线程
std::thread subscriberThread([this, &subscriber] {
while (true) {
// 在实际应用中，这里可以添加处理逻辑，如消息确认机制等
Message message = queue->pop();
subscriber.receiveMessage(message);
}
});
subscriberThread.detach();
}
void dispatchMessage(const Message& message) {
// 查询注册中心，找到订阅了该消息主题的订阅者并分发消息
const std::string& topic = message.topic;
std::vector<std::string> subscribers = subscriptions.getSubscribers(topic);
for (const auto& subscriber : subscribers) {
// 这里可以实现更复杂的消息分发策略，如消息确认机制等
queue->push(message); // 分发消息给订阅者
}
}
};
int main() {
// 创建消息路由器
MessageRouter router;
// 创建订阅者
Subscriber orderService("Order Service");
Subscriber userService("User Service");
// 订阅关系示例
Subscription subscription1 = {"product_information", "Order Service"};
Subscription subscription2 = {"product_information", "User Service"};
// 注册订阅关系
router.subscribe(orderService, "product_information");
router.subscribe(userService, "product_information");
// 启动消息路由器
router.start();
// 创建UDP套接字
int udpSocket = socket(AF_INET, SOCK_DGRAM, 0);
if (udpSocket < 0) {
std::cerr << "Error creating UDP socket" << std::endl;
return 1;
}
// 设置服务器地址和端口
struct sockaddr_in serverAddress;
serverAddress.sin_family = AF_INET;
serverAddress.sin_addr.s_addr = INADDR_ANY;
serverAddress.sin_port = htons(9000); // 选择一个合适的端口号
// 绑定套接字到服务器地址和端口
if (bind(udpSocket, (struct sockaddr*)&serverAddress, sizeof(serverAddress)) < 0) {
std::cerr << "Error binding UDP socket" << std::endl;
return 1;
}
std::cout << "Message Router is running and routing product information messages" << std::endl;
// 接收和处理消息
char buffer[1024];
struct sockaddr_in clientAddress;
socklen_t clientAddressLength = sizeof(clientAddress);
while (true) {
memset(buffer, 0, sizeof(buffer));
// 接收消息
ssize_t bytesRead = recvfrom(udpSocket, buffer, sizeof(buffer), 0, (struct sockaddr*)&clientAddress, &clientAddressLength);
if (bytesRead < 0) {
std::cerr << "Error receiving message" << std::endl;
continue;
}
// 处理消息
std::string messageContent(buffer, bytesRead);
std::string messageTopic = "unknown"; // 默认主题
// 解析消息内容，假设消息格式为 "Topic: Content"
size_t separatorPos = messageContent.find(":");
if (separatorPos != std::string::npos) {
messageTopic = messageContent.substr(0, separatorPos);
messageContent = messageContent.substr(separatorPos + 1);
}
Message message = {messageTopic, messageContent};
router.publish(message); // 将消息路由到订阅者
std::cout << "Received UDP message: " << messageContent << " from " << inet_ntoa(clientAddress.sin_addr) << ":" << ntohs(clientAddress.sin_port) << std::endl;
}
// 关闭套接字
close(udpSocket);
// 停止消息路由器
router.stop();
return 0;
}在上述代码中，创建了一个 MessageRouter 类，它维护了一个消息队列，用于存储待处理的消息。在 subscribe 方法中，订阅者会注册对特定主题的订阅关系，并为每个订阅者创建一个独立的线程来处理消息。消息路由器在 publish 方法中将接收到的消息推送到消息队列，并在单独的消息处理线程中分发消息给订阅者。
主函数中，创建了两个订阅者，订单服务和用户服务，它们都订阅了 "product_information" 主题的消息。然后，启动了消息路由器，并通过UDP（为了方便使用了UDP)套接字接收商品信息消息，将其路由给订阅者。
第五部分：总结
在编程中，消息路由是一种常见的编程范式，许多开源库和框架在其内部或作为其核心功能之一使用了消息路由。例如Apache Kafka，RabbitMQ，Apache ActiveMQ，Redis，Node.js，Spring Framework，Apache Camel，NATS,ROS1,ROS2, 百度Apollo等,它们使用消息路由来支持事件处理、消息传递和通信。消息路由是现代应用程序开发中的重要概念，它提供了一种灵活、可扩展和高性能的通信机制。学习消息路由有助于编程从业者构建更复杂、更高效和更具响应性的应用程序，适应不断变化的技术和业务要求。
待续。。。
相关推荐
-
- 驱动网卡（怎么从新驱动网卡）
-
网卡一般是指为电脑主机提供有线无线网络功能的适配器。而网卡驱动指的就是电脑连接识别这些网卡型号的桥梁。网卡只有打上了网卡驱动才能正常使用。并不是说所有的网卡一插到电脑上面就能进行数据传输了，他都需要里面芯片组的驱动文件才能支持他进行数据传输...
-
2026-01-30 00:37 liuian
- win10更新助手装系统（微软win10更新助手）
-
1、点击首页“系统升级”的按钮，给出弹框，告诉用户需要上传IMEI码才能使用升级服务。同时给出同意和取消按钮。华为手机助手2、点击同意，则进入到“系统升级”功能华为手机助手华为手机助手3、在检测界面，...
- windows11专业版密钥最新（windows11专业版激活码永久）
-
Windows11专业版的正版密钥，我们是对windows的激活所必备的工具。该密钥我们可以通过微软商城或者通过计算机的硬件供应商去购买获得。获得了windows11专业版的正版密钥后，我...
-
- 手机删过的软件恢复（手机删除过的软件怎么恢复）
-
操作步骤：1、首先，我们需要先打开手机。然后在许多图标中找到带有[文件管理]文本的图标，然后单击“文件管理”进入页面。2、进入页面后，我们将在顶部看到一行文本：手机，最新信息，文档，视频，图片，音乐，收藏，最后是我们正在寻找的[更多]，单击...
-
2026-01-29 23:55 liuian
- 一键ghost手动备份系统步骤（一键ghost 备份）
-
步骤1、首先把装有一键GHOST装系统的U盘插在电脑上，然后打开电脑马上按F2或DEL键入BIOS界面，然后就选择BOOT打USDHDD模式选择好，然后按F10键保存，电脑就会马上重启。 步骤...
- 怎么创建局域网（怎么创建局域网打游戏）
-
1、购买路由器一台。进入路由器把dhcp功能打开 2、购买一台交换机。从路由器lan端口拉出一条网线查到交换机的任意一个端口上。 3、两台以上电脑。从交换机任意端口拉出网线插到电脑上（电脑设置...
- 精灵驱动器官方下载（精灵驱动手机版下载）
-
是的。驱动精灵是一款集驱动管理和硬件检测于一体的、专业级的驱动管理和维护工具。驱动精灵为用户提供驱动备份、恢复、安装、删除、在线更新等实用功能。1、全新驱动精灵2012引擎，大幅提升硬件和驱动辨识能力...
- 一键还原系统步骤（一键还原系统有哪些）
-
1、首先需要下载安装一下Windows一键还原程序，在安装程序窗口中，点击“下一步”，弹出“用户许可协议”窗口，选择“我同意该许可协议的条款”，并点击“下一步”。 2、在弹出的“准备安装”窗口中，可...
- 电脑加速器哪个好（电脑加速器哪款好）
-
我认为pp加速器最好用，飞速土豆太懒，急速酷六根本不工作。pp加速器什么网页都加速，太任劳任怨了！以上是个人观点，具体性能请自己试。ps：我家电脑性能很好。迅游加速盒子是可以加速电脑的。因为有过之...
- 任何u盘都可以做启动盘吗（u盘必须做成启动盘才能装系统吗）
-
是的，需要注意，U盘的大小要在4G以上，最好是8G以上，因为启动盘里面需要装系统，内存小的话，不能用来安装系统。内存卡或者U盘或者移动硬盘都可以用来做启动盘安装系统。普通的U盘就可以，不过最好U盘...
- u盘怎么恢复文件（u盘文件恢复的方法）
-
开360安全卫士，点击上面的“功能大全”。点击文件恢复然后点击“数据”下的“文件恢复”功能。选择驱动接着选择需要恢复的驱动，选择接入的U盘。点击开始扫描选好就点击中间的“开始扫描”，开始扫描U盘数据。...
- 系统虚拟内存太低怎么办（系统虚拟内存占用过高什么原因）
-
1.检查系统虚拟内存使用情况，如果发现有大量的空闲内存，可以尝试释放一些不必要的进程，以释放内存空间。2.如果系统虚拟内存使用率较高，可以尝试增加系统虚拟内存的大小，以便更多的应用程序可以使用更多...
-
- 剪贴板权限设置方法（剪贴板访问权限）
-
1、首先打开iphone手机，触碰并按住单词或图像直到显示选择选项。2、其次，然后选取“拷贝”或“剪贴板”。3、勾选需要的“权限”，最后选择开启，即可完成苹果剪贴板权限设置。仅参考1.打开苹果手机设置按钮,点击【通用】。2.点击【键盘】,再...
-
2026-01-29 21:37 liuian
- 平板系统重装大师（平板重装win系统）
-
如果你的平板开不了机，但可以连接上电脑，那就能好办，楼主下载安装个平板刷机王到你的个人电脑上，然后连接你的平板，平板刷机王会自动识别你的平板，平板刷机王上有你平板的我刷机包，楼主点击下载一个，下载完成...
- 联想官网售后服务网点（联想官网售后服务热线）
-
联想3c服务中心是联想旗下的官方售后，是基于互联网O2O模式开发的全新服务平台。可以为终端用户提供多品牌手机、电脑以及其他3C类产品的维修、保养和保险服务。根据客户需求层次，联想服务针对个人及家庭客户...
- 一周热门
- 最近发表
- 标签列表
-
- python判断字典是否为空 (50)
- crontab每周一执行 (48)
- aes和des区别 (43)
- bash脚本和shell脚本的区别 (35)
- canvas库 (33)
- dataframe筛选满足条件的行 (35)
- gitlab日志 (33)
- lua xpcall (36)
- blob转json (33)
- python判断是否在列表中 (34)
- python html转pdf (36)
- 安装指定版本npm (37)
- idea搜索jar包内容 (33)
- css鼠标悬停出现隐藏的文字 (34)
- linux nacos启动命令 (33)
- gitlab 日志 (36)
- adb pull (37)
- python判断元素在不在列表里 (34)
- python 字典删除元素 (34)
- vscode切换git分支 (35)
- python bytes转16进制 (35)
- grep前后几行 (34)
- hashmap转list (35)
- c++ 字符串查找 (35)
- mysql刷新权限 (34)