一、项目源码
https://gitee.com/dongfang-weiming-0108/RabbitMQ
二、项目前言
在实际的后端开发中，尤其是分布式系统里，跨主机之间使用生产者消费者模型，是非常普遍的需求。生产者消费者模型常基于阻塞队列（一种常见的实现生产者消费者模型的数据结构，当队列为空时，从队列中获取元素的操作将被阻塞；当队列满时，从队列中存放元素的操作将被阻塞）解决忙闲不均、实现负载均衡与多主机使用。因此，我们会通常会把阻塞队列封装成一个独立的服务器程序，并赋予其更丰富的功能。这样的服务程序被称作消息队列。市面上有非常多成熟的消息队列，RabbitMQ就是其中之一。
本项目旨在仿照RabbitMQ实现一个简易的消息队列，消息发布客户端（生产者）把消息推送到消息队列服务器，服务器再根据消息订阅客户端（消费者）的订阅情况，把消息推送给相关订阅者。（PS：由于是一个简化版的消息队列组件，所以功能并不如真正的RabbitMQ一般完善，还有许多地方有待扩展）
项目运行步骤
1、启动服务端
2、订阅客户端启动（客户运行时还需要添加想要订阅的队列名）
需要先启动订阅客户端，订阅想要订阅的队列名，如果先启动发布客户端，消息已经发布，会出现消息丢失的情况
3、启动发布客户端（使用后台接口进行发布消息，此处使用主体交换匹配模式为例）
4、订阅客户端收到消息
queue1由于不匹配，未接收到消息
queue2匹配成功接收到消息
三、开发环境与技术选型
1、开发环境
CentOS 7云服务器，VsCode，g++/gdb，Makefile
2、技术选型
开发主语言C++，Protobuf二进制序列化框架，自定义应用层协议+Muduo库，SQLite3源数据信息数据库，Gtest单元测试框架
2.1 Protobuf的简要介绍
2.1.1 什么是Protobuf
Protobuf（全称Protocol Buffer）是谷歌公司开发的一种对数据结构进行序列化和反序列化的框架，具有以下优点：
1、和语言与平台无关： ProtoBuf 支持 C++、Java、Python 等多种语言；也支持多个平台
2、效率高：比XML（可扩展标记语言）更小、更快、也更简单
3、扩展性强，兼容性好：可以在更新数据结构的同时，不影响和破坏原有的旧程序
2.2.2 Protubuf的使用流程
1、编写.proto后缀的文件，描述想要定义的结构化对象，并描述对象中有什么样的成员，每个成员的属性是什么；目的是为了定义结构对象及属性内容（定义过程中，需要声明语法版本，常用proto3语法；声明命名空间名称；然后再定义结构化对象描述）
.proto文件编写示意图
2、使用protoc编译器编译.proto文件（protoc --cpp_out=所在目录 xx.proto），编译后会生成一系列接口代码，存放在.cc源文件（定义实现了结构化对象数据的访问、操作、序列化、反序列化）和.h头文件（定义了描述的数据结构对象类）中。
3、把编译生成的头文件包含进我们的代码中，依赖生成的接口，即可实现设置和获取.proto文件中定义的字段，以及对message对象进行序列化和反序列化
Protobuf文件关系示意图
2.2 Muduo库的简要介绍
2.2.1 什么是Muduo库
Muduo库是由陈硕大佬开发的一个基于非阻塞IO和事件驱动的C++高并发TCP网络编程库，是一款基于主从Reactor模型的高性能服务器框架。
2.2.2 主从Reactor模型的简单介绍
所谓Reactor模型，实际上是一种基于事件触发（即基于epoll多路转接进行IO事件监控）的模型。而主从Reactor模型，就是把IO事件监控进行了更进一步的层次划分，划分成主Reactor线程和从属Reactor线程。
主Reactor线程：只对新建连接事件进行监控，保证其不受IO阻塞的影响，实现新建连接的高效获取。
从属Reactor线程：针对普通连接进行IO事件的操作与业务处理，从属Reactor线程可以有多个，负载均衡式地处理IO事件。
因此主从Reactor模型是一个多执行流的并发模式，每一个事件监控占据一个线程，从而导致其内部必然有一个Reactor线程池。
主从Reactor模型示意图
2.2.3 Muduo库的几个常用接口
1、muduo::net::TcpServer类的简单介绍
typedef std::shared_ptr<TcpConnection> TcpConnectionPtr;
typedef std::function<void(const TcpConnectionPtr &)> ConnectionCallback;
typedef std::function<void(const TcpConnectionPtr &, // 连接
Buffer *, // 消息
Timestamp)> // 时间
MessageCallback; // 回调函数的固定格式
class InetAddress : public muduo::copyable
{
public:
InetAddress(StringArg ip, uint16_t port, bool ipv6 = false);
};
class TcpServer : noncopyable
{
public:
enum Option
{
kNoReusePort,
kReusePort,
};
TcpServer(EventLoop *loop, // 事件监控循环处理
const InetAddress &listenAddr, // 服务器所要监听绑定的地址信息（IP地址+端口号）
const string &nameArg, // 服务器名称
Option option = kNoReusePort); // 是否启用端口复用（一个链接如果主动断开，会进入timewait状态，即等待阶段，此时不能立即使用它，如果开启端口复用，可以立刻使用处于等待阶段的端口）
void setThreadNum(int numThreads); // 设置丛书Reactor线程的数量
void start(); // 启动服务器、开始套接字监听等
/// 当⼀个新连接建⽴成功的时候被调⽤（用户自己实现，希望新链接建立成功后做什么的函数，调用此接口，把该函数设置为回调函数，调用链接时将自动调用该函数）
void setConnectionCallback(const ConnectionCallback &cb)
{
connectionCallback_ = cb;
}
/// 消息的业务处理回调函数---这是收到新连接消息的时候被调⽤的函数（同样由用户自己决定）
void setMessageCallback(const MessageCallback &cb)
{
messageCallback_ = cb;
}
};
2、muduo::net::EventLoop类（即事件循环类）的简单介绍
class EventLoop : noncopyable
{
public:
void loop(); // 内部就是一个事件监控，触发事件后进行IO处理的死循环（必须在创建对象的同一线程中调用）
void quit(); // 退出循环，如果通过原始指针调用，不是100%线程安全，为了100%的安全性，最好通过shared_ptr<EventLoop>调用。
TimerId runAt(Timestamp time, TimerCallback cb);
TimerId runAfter(double delay, TimerCallback cb); // 在@c延迟秒后运行回调。可以在其他线程中安全调用
TimerId runEvery(double interval, TimerCallback cb); // 每隔@c延迟秒运行一次回调。可以在其他线程中安全调用
void cancel(TimerId timerId); // 取消计时器
private:
std::atomic<bool> quit_;
std::unique_ptr<Poller> poller_;
mutable MutexLock mutex_;
std::vector<Functor> pendingFunctors_ GUARDED_BY(mutex_);
};
3、muduo::net::TcpConnection类的简单介绍
class TcpConnection : noncopyable, public std::enable_shared_from_this<TcpConnection>
{
public:
// Constructs a TcpConnection with a connected sockfd
// User should not create this object.
TcpConnection(EventLoop *loop,
const string &name,
int sockfd,
const InetAddress &localAddr,
const InetAddress &peerAddr);
bool connected() const { return state_ == kConnected; } // 判断当前是否处于连接状态
bool disconnected() const { return state_ == kDisconnected; } // 判断当前是否处于连接断开状态
void send(string &&message); // C++11（用的最多）
void send(const void *message, int len);
void send(const StringPiece &message);
// void send(Buffer&& message); // C++11
void send(Buffer *message); // 交换数据
void shutdown(); // 不是线程安全，没有并发调用
void setContext(const boost::any &context)
{
context_ = context;
}
const boost::any &getContext() const
{
return context_;
}
boost::any *getMutableContext()
{
return &context_;
}
void setConnectionCallback(const ConnectionCallback &cb)
{
connectionCallback_ = cb;
}
void setMessageCallback(const MessageCallback &cb)
{
messageCallback_ = cb;
}
private:
enum StateE
{
kDisconnected,
kConnecting,
kConnected,
kDisconnecting
};
EventLoop *loop_;
ConnectionCallback connectionCallback_;
MessageCallback messageCallback_;
WriteCompleteCallback writeCompleteCallback_;
boost::any context_;
};
4、muduo::net::TcpClient类的简单介绍
class TcpClient : noncopyable
{
public:
// TcpClient(EventLoop* loop);
// TcpClient(EventLoop* loop, const string& host, uint16_t port);
TcpClient(EventLoop *loop, const InetAddress &serverAddr, const string &nameArg);
~TcpClient(); // force out-line dtor, for std::unique_ptr members.
void connect(); // 连接服务器（非阻塞接口，即发起请求后会立即返回，但此时连接不一定成功）
void disconnect(); // 关闭连接
void stop();
// 获取客户端对应的通信连接Connection对象的接⼝，发起connect后，有可能还没有连接建立成功，此时将获取空连接
TcpConnectionPtr connection() const
{
MutexLockGuard lock(mutex_);
return connection_;
}
/// 连接服务器成功时的回调函数
void setConnectionCallback(ConnectionCallback cb)
{
connectionCallback_ = std::move(cb);
}
/// 收到服务器发送的消息时的回调函数
void setMessageCallback(MessageCallback cb)
{
messageCallback_ = std::move(cb);
}
private:
EventLoop *loop_;
ConnectionCallback connectionCallback_;
MessageCallback messageCallback_;
WriteCompleteCallback writeCompleteCallback_;
TcpConnectionPtr connection_ GUARDED_BY(mutex_);
};
/*
需要注意的是，因为muduo库不管是服务端还是客户端都是异步操作，对于客⼾端来说如果我们在连接还没有完全建⽴成功的时候发送数据，这是不被允许的。
因此我们可以使⽤内置的CountDownLatch类进⾏同步控制
*/
class CountDownLatch : noncopyable
{
public:
explicit CountDownLatch(int count);
void wait()
{
MutexLockGuard lock(mutex_);
while (count_ > 0)
{
condition_.wait();
}
}
void countDown()
{
MutexLockGuard lock(mutex_);
--count_;
if (count_ == 0)
{
condition_.notifyAll();
}
}
int getCount() const;
private:
mutable MutexLock mutex_;
Condition condition_ GUARDED_BY(mutex_);
int count_ GUARDED_BY(mutex_);
};
2.3 Muduo库内部实现的几个基于Protobuf的接口类
2.3.1 ProtobufCodec类
ProtobufCodec类是muduo库中对于protobuf协议的处理类，其内部实现了onMessage的回调接口，对于接收到的数据进行给予protobuf协议的请求处理，然后把解析出的信息存放到请求的protobuf请求类对象中，然后最后调用设置进去的消息处理回调函数进行对应请求的处理
/*muduo-master/examples/protobuf/codec.h*/
typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
class ProtobufCodec : muduo::noncopyable
{
public:
enum ErrorCode
{
kNoError = 0,
kInvalidLength,
kCheckSumError,
kInvalidNameLen,
kUnknownMessageType,
kParseError,
};
typedef std::function<void(const muduo::net::TcpConnectionPtr &, const MessagePtr &, muduo::Timestamp)> ProtobufMessageCallback;
// 这⾥的messageCb是针对protobuf请求进⾏处理的函数，它声明在dispatcher.h中的
ProtobufDispatcher类 explicit ProtobufCodec(const ProtobufMessageCallback &messageCb)
: messageCallback_(messageCb), // 这就是设置的请求处理回调函数
errorCallback_(defaultErrorCallback)
{
}
// 它的功能就是接收消息，进⾏解析，得到了proto中定义的请求后调⽤设置的messageCallback_进⾏处理
void onMessage(const muduo::net::TcpConnectionPtr &conn,
muduo::net::Buffer *buf,
muduo::Timestamp receiveTime);
// 通过conn对象发送响应的接⼝
void send(const muduo::net::TcpConnectionPtr &conn,
const google::protobuf::Message &message)
{
// FIXME: serialize to TcpConnection::outputBuffer()
muduo::net::Buffer buf;
fillEmptyBuffer(&buf, message);
conn->send(&buf);
}
static const muduo::string &errorCodeToString(ErrorCode errorCode);
static void fillEmptyBuffer(muduo::net::Buffer *buf, const google::protobuf::Message &message);
static google::protobuf::Message *createMessage(const std::string type_name);
static MessagePtr parse(const char *buf, int len, ErrorCode *errorCode);
private:
static void defaultErrorCallback(const muduo::net::TcpConnectionPtr &,
muduo::net::Buffer *,
muduo::Timestamp,
ErrorCode);
ProtobufMessageCallback messageCallback_;
ErrorCallback errorCallback_;
const static int kHeaderLen = sizeof(int32_t);
const static int kMinMessageLen = 2 * kHeaderLen + 2; // nameLen + typeName +checkSum
const static int kMaxMessageLen = 64 * 1024 * 1024; // same as codec_stream.hkDefaultTotalBytesLimit
};
2.3.2 ProtobufDispatcher类
ProtobufDispatcher类是一个比较重要的类，是一个protobuf请求的分发处理类，用户在使用的时候，在这个类对象中注册，哪个请求应该用那个业务函数进行处理。
其内部的onProtobufMessage接口，就是给上边ProtobufCodec::messageCallback_设置的回调函数，相当于ProtobufCodec中onMessage接口会设置给服务器，作为消息回调函数，其内部对于接收到的数据进行基于protobuf协议的解析，得到请求后通过ProtobufDispatcher::onProtobufMessage接口进行请求分发处理，也就是确定当前请求应该用哪一个注册过的业务函数进行处理。
typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
class Callback : muduo::noncopyable
{
public:
virtual ~Callback() = default;
virtual void onMessage(const muduo::net::TcpConnectionPtr &,
const MessagePtr &message,
muduo::Timestamp) const = 0;
};
// 这是⼀个对函数接⼝进⾏⼆次封装⽣成⼀个统⼀类型对象的类
template <typename T>
class CallbackT : public Callback
{
static_assert(std::is_base_of<google::protobuf::Message, T>::value,
"T must be derived from gpb::Message.");
public:
typedef std::function<void(const muduo::net::TcpConnectionPtr &,
const std::shared_ptr<T> &message,
muduo::Timestamp)>
ProtobufMessageTCallback;
CallbackT(const ProtobufMessageTCallback &callback)
: callback_(callback)
{
}
void onMessage(const muduo::net::TcpConnectionPtr &conn,
const MessagePtr &message,
muduo::Timestamp receiveTime) const override
{
std::shared_ptr<T> concrete = muduo::down_pointer_cast<T>(message);
assert(concrete != NULL);
callback_(conn, concrete, receiveTime);
}
private:
ProtobufMessageTCallback callback_;
};
// 这是⼀个protobuf请求分发器类，需要⽤⼾注册不同请求的不同处理函数，
// 注册完毕后，服务器收到指定请求就会使⽤对应接⼝进⾏处理
class ProtobufDispatcher
{
public:
typedef std::function<void(const muduo::net::TcpConnectionPtr &,
const MessagePtr &message,
muduo::Timestamp)>
ProtobufMessageCallback;
// 构造对象时需要传⼊⼀个默认的业务处理函数，以便于找不到对应请求的处理函数时调⽤。
explicit ProtobufDispatcher(const ProtobufMessageCallback &defaultCb)
: defaultCallback_(defaultCb)
{
}
// 这个是⼈家实现的针对proto中定义的类型请求进⾏处理的函数，内部会调⽤我们⾃⼰传⼊的业务处理函数
void onProtobufMessage(const muduo::net::TcpConnectionPtr &conn,
const MessagePtr &message,
muduo::Timestamp receiveTime) const
{
CallbackMap::const_iterator it = callbacks_.find(message->GetDescriptor());
if (it != callbacks_.end())
{
it->second->onMessage(conn, message, receiveTime);
}
else
{
defaultCallback_(conn, message, receiveTime);
}
}
/*
这个接⼝⾮常巧妙，基于proto中的请求类型将我们⾃⼰的业务处理函数与对应的请求给关联起来了
相当于通过这个成员变量中的CallbackMap能够知道收到什么请求后应该⽤什么处理函数进⾏处理
简单理解就是注册针对哪种请求--应该⽤哪个我们⾃⼰的函数进⾏处理的映射关系
但是我们⾃⼰实现的函数中，参数类型都是不⼀样的⽐如翻译有翻译的请求类型，加法有加法请求类型
⽽map需要统⼀的类型，这样就不好整了，所以⽤CallbackT对我们传⼊的接⼝进⾏了⼆次封装。
*/
template <typename T>
void registerMessageCallback(const typename CallbackT<T>::ProtobufMessageTCallback &callback)
{
std::shared_ptr<CallbackT<T>> pd(new CallbackT<T>(callback));
callbacks_[T::descriptor()] = pd;
}
private:
typedef std::map<const google::protobuf::Descriptor *, std::shared_ptr<Callback>> CallbackMap;
CallbackMap callbacks_;
ProtobufMessageCallback defaultCallback_;
};
2.4 重要元素：应用层协议
Muduo库自定义应用层协议示意图
protobuf根据我们的proto文件所生成的代码中，会生成对应类型的类（eg：TranslateRequest对应生成一个TranslateRequest类） 实际上，protobuf还做了更多的事情——每隔对应的类中都包含有一个描述结构的指针：
static const ::google::protobuf::Descriptor* descriptor();
这个描述结构非常重要，其内部可以获取到当前对应类的类型名称，以及各项成员的名称，通过这些名称+协议中的typename字段，就可以实现完美的对应关系。
协议处理流程图
2.5 SQLite3的简要介绍
2.5.1 什么是SQLite
SQLite是一个进程内的轻量级数据库，是一种实现了自给自足的、无服务器的、零配置的、事务性的SQL数据库引擎。
与其他数据库类似，SQLite引擎也不是一个独立的进程，其可以按照应用程序的不同需求进行静态 / 动态连接，直接访问其存储文件。
与其他数据库不同的是，SQLite是一个零配置的数据库，不需要在系统中配置。
2.5.2 本项目为什么要使用SQLite
1、SQLite是无服务器的，不需要一个单独的服务器进程 / 操作系统。
2、SQLite不需要配置，自给自足，不需要任何外部依赖。
3、一个完整的SQLite数据库存储在一个单一的、跨平台的磁盘文件上。
4、SQLite是一个轻量级数据库，省略可选功能配置时大小小于250KiB，即便完全配置，占据空间也不会大于400KIB。
5、SQLite事务是完全兼容ACID（数据库管理系统必须具备的四个特性：原子性、一致性、隔离性、持久性）的，允许从多个进程 / 线程安全访问。
6、SQLite支持SQL92（SQL2）标准的大多数数据库查询语言。
7、SQLite使用ANSI-C编写，提供的API简单且利于使用。
8、SQLite可以在UNIX（Linux、Mac OS-X，Android、iOS）和Windows（Win32、WinCE、WinRT）等多个平台运行，兼容性好。
SQLite3 C/C++ API介绍
SQLiete3官方文档链接：List Of SQLite Functions
C/C++ API是SQLite3数据库的一个客户端，提供使用C/C++操作数据库的方法
2.5.3 SQLite3的操作流程
1、查看当前数据库在编译阶段是否启动了线程安全（默认启用）
int sqlite3_threadsafe()； // 返回值：0-未启⽤； 1-启⽤
需要注意的是SQLite3有三种安全等级：非线程安全模式（*效率最高*）、线程安全模式（*不同的连接在不同的线程 / 进程是安全的，即**一个句柄不能用于多线程之间***）、串行化模式*（可以在不同的线程 / 进程间使用同一个句柄）*
2、创建 / 打开数据库文件，并返回操作句柄
int sqlite3_open(const char *filename, sqlite3 **ppDb) // 成功返回SQLITE_OK
// 若在编译阶段启动了线程安全，则在程序运⾏阶段可以通过参数选择线程安全等级，
int sqlite3_open_v2(const char *filename, sqlite3 **ppDb, int flags, const char *zVfs );
/*
flags即文件打开方式，有四种
1、SQLITE_OPEN_READWRITE -- 以可读可写⽅式打开数据库⽂件
2、SQLITE_OPEN_CREATE -- 不存在数据库⽂件则创建
3、SQLITE_OPEN_NOMUTEX--多线程模式，只要不同的线程使⽤不同的连接即可保证线程安全
4、SQLITE_OPEN_FULLMUTEX--串⾏化模式
3和4都是线程安全的设置方式；设置成功同样返回SQLITE_OK
*/
3、执行语句
int sqlite3_exec(sqlite3*, char *sql, int (*callback)(void*,int,char**,char**), void* arg,char **err)
/*
返回值：SQLITE_OK表示成功
int (*callback)(void*,int,char**,char**)的介绍：
void* : 是设置的在回调时传入的arg参数
int：⼀⾏中数据的列数
char**：存储⼀⾏数据的字符指针数组
char**：每⼀列的字段名称
这个回调函数有个int返回值，成功处理的情况下必须返回0，返回⾮0会触发ABORT退出程序
*/
4、销毁句柄
int sqlite3_close(sqlite3* db); // 成功返回SQLITE_OK
int sqlite3_close_v2(sqlite3*); // 推荐使⽤--⽆论如何都会返回SQLITE_OK
const char *sqlite3_errmsg(sqlite3* db); // 获取错误信息
2.6 GTest的简要介绍
2.6.1 GTest是什么
GTtest是一个由Google公司发布的、跨平台的C++单元测试框架；其目的是为了可以在不同平台上上编写C++单元测试程序。GTest提供了丰富的断言、致命和非致命判断、参数化等测试功能。
2.6.2 GTest的使用
1、TEST宏
TEST(test_case_name, test_name)
TEST_F(test_fixture,test_name)
TEST：主要用来创建一个简单测试，它定义了一个测试函数，在这个函数中可以使用任何C++代码，使用框架提供的断言进行检查
TEST_F：主要用来进行多样的测试，适用于在多个测试场景，需要相同的数据配置的情况，即相同的数据测不同的行为
2、断言
GTest中的断言宏可以分为两类（都必须在单元测试宏函数中使用）：ASSERT_系列（如果当前点检测失败，则退出当前函数，本项目使用的断言宏就是此系列）；EXPECT_系列（如果当前点检测失败，继续往下执行）
常用的断言：
// bool值检查
ASSERT_TRUE(参数)，期待结果是true（是否为真）
ASSERT_FALSE(参数)，期待结果是false（是否为假）
// 数值型数据检查
ASSERT_EQ(参数1，参数2)，传⼊的是需要⽐较的两个数 equal
ASSERT_NE(参数1，参数2)，not equal，不等于才返回true
ASSERT_LT(参数1，参数2)，less than，⼩于才返回true
ASSERT_GT(参数1，参数2)，greater than，⼤于才返回true
ASSERT_LE(参数1，参数2)，less equal，⼩于等于才返回true
ASSERT_GE(参数1，参数2)，greater equal，⼤于等于才返回true
PS：如果对自动输出的错误信息不满意，可以通过operator<<在失败的时候打印自定义日志
2.6.3 事件机制
GTest中的事件机制，指的是在测试前和测试后提供的让用户自行添加操作的机制，该机制同时也可以让同一测试套件下的测试用例共享数据。
事件机制的最大好处就是可以为我们各个用例提前准备好测试环境，并在测试完毕后销毁；如果我们有一段代码需要进行不同方法的测试，可以通过事件机制在每个测试用例进行之前初始化测试环境和数据，并在测试完毕后清理测试所造成的影响。
Gtest框架下，事件的结构层次
测试程序：一个测试程序只能有一个main函数，也可以说是一个可执行程序是一个测试程序。该级别的事件机制，是在测试程序的开始和结束时执行。
测试套件：代表一个测试用例的集合体，可以理解为一个测试环境，可以在单元测试前进行测试环境的初始化，测试完毕后进行测试环境的清理。该级别的事件机制，是在整体的测试用例的开始和结束时执行。（分为全局测试套件和用例测试套件）
（PS：测试中，可以有多个测试套件，每个测试套件包含一组测试，每组测试中包含一组单元测试）
全局测试套件：在整体测试中只会初始化一次环境，在所有测试用例执行完毕后才会清理环境。
用例测试套件：每次单元测试时，都会重新初始化测试环境，执行完毕就清理环境。
测试用例：该级别的事件机制在每个测试用例开始和结束时都执行。
GTest提供的三种常见事件
1、全局事件
全局事件针对整个测试程序。实现全局的事件机制，需要创建⼀个自己的类，然后继承testing::Environment类，再分别实现成员函数 SetUp 和 TearDown，同时在main函数内进行调用 testing::AddGlobalTestEnvironment(new MyEnvironment); 函数添加全局的事件机制
运行示例：
#include <iostream>
#include <gtest/gtest.h>
// 全局事件:针对整个测试程序,提供全局事件机制,能够在测试之前配置测试环境数据,测试完毕后清理数据
// 先定义环境类,通过继承testing::Environment的派⽣类来完成
// 重写的虚函数接⼝SetUp会在测试之前被调⽤; TearDown会在测试完毕后调⽤
std::unordered_map<std::string, std::string> dict;
class HashTestEnv : public testing::Environment
{
public:
virtual void SetUp() override
{
std::cout << "测试前:提前准备数据!!\n";
dict.insert(std::make_pair("Hello", "你好"));
dict.insert(std::make_pair("hello", "你好"));
dict.insert(std::make_pair("雷吼", "你好"));
}
virtual void TearDown() override
{
std::cout << "测试结束后:清理数据!!\n";
dict.clear();
}
};
TEST(hash_case_test, find_test)
{
auto it = dict.find("hello");
ASSERT_NE(it, dict.end());
}
TEST(hash_case_test, size_test)
{
ASSERT_GT(dict.size(), 0);
}
int main(int argc, char *argv[]) 30
{
testing::AddGlobalTestEnvironment(new HashTestEnv);
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
运行结果：
2、TestSuite事件
TestSuite事件针对⼀个个测试套件。测试套件的事件机制我们同样需要去创建⼀个类，继承自
testing::Test，实现两个静态函数 SetUpTestCase 和 TearDownTestCase，测试套件的事件机制不需要像全局事件机制⼀样在 main 注册，而是需要将我们平时使用的TEST宏改为TEST_F宏。
1、SetUpTestCase() 函数是在测试套件第⼀个测试用例开始前执行
2、TearDownTestCase() 函数是在测试套件最后⼀个测试用例结束后执行
3、需要注意TEST_F的第⼀个参数是我们创建的类名，也就是当前测试套件的名称，这样在
TEST_F宏的测试套件中就可以访问类中的成员了。
测试示例：
#include <iostream>
#include <unordered_map>
#include <gtest/gtest.h>
// TestSuite:测试套件/集合进⾏单元测试,即,将多个相关测试归⼊⼀组的⽅式进⾏测试,为这组测试⽤例进⾏环境配置和清理
// 概念: 对⼀个功能的验证往往需要很多测试⽤例，测试套件就是针对⼀组相关测试⽤例进⾏环境配置的事件机制
// ⽤法: 先定义环境类,继承于 testing::Test 基类, 重写两个静态函数SetUpTestCase TearDownTestCase进⾏环境的配置和清理
class HashTestEnv1 : public testing::Test
{
public:
static void SetUpTestCase()
{
std::cout << "环境1第⼀个TEST之前调⽤\n";
}
static void TearDownTestCase()
{
std::cout << "环境1最后⼀个TEST之后调⽤\n";
}
public:
std::unordered_map<std::string, std::string> dict;
};
// 注意,测试套件使⽤的不是TEST了,⽽是TEST_F, ⽽第⼀个参数名称就是测试套件环境类名称
// main函数中不需要再注册环境了,⽽是在TEST_F中可以直接访问类的成员变量和成员函数
TEST_F(HashTestEnv1, insert_test)
{
std::cout << "环境1,中间insert测试\n";
dict.insert(std::make_pair("Hello", "你好"));
dict.insert(std::make_pair("hello", "你好"));
dict.insert(std::make_pair("雷吼", "你好"));
auto it = dict.find("hello");
ASSERT_NE(it, dict.end());
}
TEST_F(HashTestEnv1, sizeof)
{
std::cout << "环境1,中间size测试\n";
ASSERT_GT(dict.size(), 0);
}
int main(int argc, char *argv[])
{
testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
运行结果
从上例中可以看出，TestCase事件的好处是把数据与测试结合到同一个测试环境类TEST_F中了，和外界的耦合度更低，代码更加清晰。
同时我们也会发现，第二个测试用例（dict.size() ＞0时返回true）失败了，这就涉及到了TestCase事件的机制——每个测试用例的环境都是相互独立，互不影响的。
TestCase事件的机制
TestCase事件是针对一个个测试用例的，测试用例的事件机制的创建和测试套件基本相同，不同的地方是，测试用例实现的两个函数SetUp与TearDown。
SetUp()函数在每一个测试用例的开始前执行，TearDown()函数在每一个测试用例的结束后执行。
也就是说，在TestSuite/TestCase事件中，每个测试用例虽然它们共用同一个事件环境类，可以访问其中的资源，但是本质上每个测试用例的环境都是独立的，这样我们就不用担心不同测试用例之间互相会有数据上的影响了，保证所有的测试用例都可以使用相同的测试环境进行测试。
四、需求分析
1、核心概念
消息发布客户端（生产者）把消息推送到服务器，服务器再根据订阅客户端（消费者）的订阅情况，把消息推送给指定的消费者。（N个生产者——N个消费者）
项目核心流程图
图中，最核心的部分即为Broker Server（消息队列服务器），负责消息的存储和转发。
根据AMQP模型（Advanced Message Queuing Protocol 高级消息队列协议，一个提供统一消息服务的应用层标准高级消息队列协议，为了面向消息的中间件而设计，使得遵从该规范的客户端应用和消息中间件服务器的全功能互动操作成为可能），消息中间件服务器Broker Server中存在以下概念：
1、消息（Message）：传递的内容
2、队列（Queue）：真正用来存储消息的部分，每个消费者自己来决定从哪个Queue上读取消息
3、交换机（Exchange）：用来绑定多个队列，转发消息的部分；生产者会先把消息发送到Broker Server的Exchange，再根据不同的匹配规则（直接交换、广播交换、主题交换），把消息转发给不同的Queue
4、绑定（Binding）：存储Exchange和Queue之间关联关系的部分，交换机和队列之间可以理解为“多对多”（即一个Exchange可以绑定多个 Queue，可以向多个 Queue 中转发消息；一个Queue也可以被多个Exchange绑定，即一个Queue的消息可以来自多个Exchange）的关系，用一个关联表就可以把这两个概念联系起来
5、虚拟机（VirtualHost）：可以视为交换机、队列、绑定三者的集合，存储消息交换过程中所涉及的绝大部分数据，是客户端操作的基础。一个Broker Server上可以存在多个虚拟机
上述的数据结构都有持久化需求，既要在内存中存储（方便使用），也要在硬盘中存储（保证重启后数据不丢失），以确保数据的持久性。
Broker Server的分解图
生产者和消费者之间的具体通信流程
2、核心接口
对于Server Broker而言，需要实现以下接口，通过这些接口来实现消息队列的基本功能；Producer 和 Consumer 则通过网络的方式远程调用这些接口，实现生产者消费者模型
1、创建 / 声明交换机（DeclareExchange）
2、销毁交换机（DeleteExchange）
3、创建 / 声明队列（DeclareQueue）
4、销毁队列（DeleteQueue）
5、队列绑定（QueueBinding）
6、队列解绑（QueueUnBinding）
7、订阅消息（BasicConsumeMessage）
8、确认消息（BasicAckMessage）
9、取消订阅（BasicCancel）
3、网络通信
生产者和消费者都是客户端程序，而Broker Server则是服务器，需要通过网络进行通信。因此，在网络通信的过程中，客户端部分需要提供对应的接口，实现对服务器的操作。
客户端提供的接口
1、创建 / 声明交换机（DeclareExchange）
2、销毁交换机（DeleteExchange）
3、创建 / 声明队列（DeclareQueue）
4、销毁队列（DeleteQueue）
5、队列绑定（QueueBinding）
6、队列解绑（QueueUnBinding）
7、订阅消息（BasicConsumeMessage）
8、确认消息（BasicAckMessage）
9、取消订阅（BasicCancel）
10、新建连接（NewConnection）
11、关闭连接（DeleteConnection）
12、打开 / 创建信道（OpenChannel）
13、关闭信道（CloseChannel）
可以看到客户端的接口，就是在Broker Server的接口基础上，多了对连接和信道的操作。
每一个连接，就对应着一个TCP通信连接，一个连接之中，可以包含多个信道。****信道与信道之间数据相互独立，互不影响。之所以在多出信道这个概念，是为了能够更好地复用TCP连接，达到长连接的效果，避免频繁的创建关闭连接。
4、消息应答
被消费完毕的消息，需要进行应答。应答模式共分为两种：
1、自动应答：消费者只要消费了消息，就算应答完毕，Broker Server便直接删除此消息。
2、手动应答：消费者手动调用应答接口，Broker只有在收到应答请求后，才真正删除此消息（手动应答的目的是为了保证消息确实被消费者处理成功了，在一些对于数据可靠性要求较高的场景中比较常见）
五、公共模块的实现
日志工具
为了便于了解项目中的错误及运行情况，编写一个日志工具模块Log.hpp，进行简单的日志打印
#pragma once
// 日志
#include <iostream>
#include <fstream>
#include <string>
#include <cstdarg>
#include <ctime>
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <pthread.h>
using namespace std;
// 想把日志打印在哪里
enum
{
screen = 10, // 打印到显示器
onefile, // 往单个文件里打
allfile, // 往多个文件中打印
};
// 枚举日志等级
enum
{
debug = 0, // 日志等级从0开始
info, // 常规信息
warning, // 不影响程序执行，但是报警
error, // 程序报错但不致命，需要解决，但是可以继续往后执行
fatal, // 服务器出现致命错误，需要直接终止
};
// 把日志等级以字符串的形式进行显示
string LevelToString(int level)
{
switch (level)
{
case debug:
return "debug";
case info:
return "info";
case warning:
return "warning";
case error:
return "error";
case fatal:
return "fatal";
default:
return "未知等级！";
}
}
const int defaultstyle = screen; // 默认往显示器上打
const std::string default_filename = "log."; // 默认文件名
const std::string logdir = "log"; // 默认文件目录
class Log
{
public:
// 构造函数
Log() : style(defaultstyle), filename(default_filename) // 初始化列表构造
{
mkdir(logdir.c_str(), 0775); // 在当前目录下创建一个目录，用于专门存储日志
};
// 把时间戳转换成当前时间
string TimeStampExchange()
{
time_t TimeStamp = time(nullptr); // 获取当前时间戳
struct tm *LocalTime = localtime(&TimeStamp); // 把时间戳转换成当前时间
char CurTime[1024];
snprintf(CurTime, sizeof(CurTime), "%d-%d-%d %d:%d:%d", LocalTime->tm_year + 1900, LocalTime->tm_mon + 1, LocalTime->tm_mday, LocalTime->tm_hour, LocalTime->tm_min, LocalTime->tm_sec);
return CurTime;
}
// 用于设置日志打印到哪里
void WhereWrite(int sty)
{
style = sty;
}
void WriteLogToOneFile(const std::string &logname, const std::string &message)
{
umask(0);
int fd = open(logname.c_str(), O_CREAT | O_WRONLY | O_APPEND, 0666); // 打开文件
if (fd < 0)
return;
write(fd, message.c_str(), message.size());
close(fd);
}
// 把日志写入文件中
void WriteLogToClassFile(const string &levelstr, const string &message)
{
std::string logname = logdir;
logname += "/"; // 拼写路径
logname += filename; // 去指定路径下写
logname += levelstr;
WriteLogToOneFile(logname, message);
}
// 把日志以写到哪里，传入日志等级和日志消息
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
void WhereWriteLog(const string &levelstr, const string &message)
{
pthread_mutex_lock(&lock);
switch (style)
{
case screen:
cout << message;
break;
case onefile:
WriteLogToClassFile("all", message);
break;
case allfile:
WriteLogToClassFile(levelstr, message);
break;
default:
break;
}
pthread_mutex_unlock(&lock);
}
void LogMessage(int level, const char *format, ...) // 类C的日志主函数，传日志等级，打印格式和可变参数（可变参数不能在函数列表中单独出现，至少要有一个合法参数，可变参数根据距离最近的合法参数的地址寻找）进行打印
{
char message[1024]; // 把信息打印到其中
va_list args; // 可变参数的指针
va_start(args, format); // 可变参数初始化，传入args和距离...最近的参数，arg会根据距离其最近的参数，使其指向可变参数部分
// args指向了可变参数部分
vsnprintf(message, sizeof(message), format, args); // 把可变参数args以format格式（如"%d %s %f"等类C格式）打印到message中
va_end(args); // 令args = nullptr，直接为空
string localtime = TimeStampExchange();
printf("[%s][%s]%s", localtime.c_str(), LevelToString(level).c_str(), message); // 由于以C形式打印，故C++中的string需要调用c_str函数把C++字符串形式转成C的字符串形式
}
~Log() {} // 析构函数
private:
int style; // 把文件写到哪里
string filename; // 把日志写入文件
};
Log lg;
// 单独列出来一个类用来启动日志
class Conf
{
public:
Conf()
{
lg.WhereWrite(screen);
}
~Conf()
{
}
};
Conf conf;
1、实用工具
实用工具模块Helper.hpp主要完成项目中各个部分都要用到的辅助功能代码——SQLite基础操作类、字符串操作类、UUID生成器类、文件基础操作类；具体内容在代码中有所体现
// 辅助工具类
#pragma once
#include <iostream>
#include <string>
#include <vector>
#include <random>
#include <sstream>
#include <fstream>
#include <iomanip>
#include <atomic>
#include <sqlite3.h>
#include "Log.hpp"
#include "boost/algorithm/string.hpp"
using namespace std;
namespace ns_helper
{
// SQLite基础操作类
/*
封装实现一个SQLiteHelper类，提供简单的sqlite数据库操作接口，完成数据的基础增删改查操作。
1. 创建/打开数据库文件
2. 针对打开的数据库执行操作
1. 表的操作（表的增删查改等）
2. 数据的操作
3. 关闭数据库
*/
class SQLiteHelper
{
public:
typedef int (*SqliteCallback)(void *, int, char **, char **); // 回调函数
SQLiteHelper(const string &dbfile) : _dbfile(dbfile), _handler(nullptr)
{
}
// 打开数据库（传入线程安全等级）
bool open(int SafeLevel = SQLITE_OPEN_FULLMUTEX)
{
// 传入参数：数据库文件名，句柄，文件打开方式，nullptr
int res = sqlite3_open_v2(_dbfile.c_str(), &_handler, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SafeLevel, nullptr);
if (res != SQLITE_OK)
{
lg.LogMessage(error, "数据库文件打开失败！失败原因：%s\n", sqlite3_errmsg(_handler));
return false;
}
return true;
}
// 执行语句（传入sql语句，结果回调函数（一般查询时候用），最后一个参数是传入回调函数中的）
bool exec(const std::string &sql, SqliteCallback cb, void *arg)
{
int res = sqlite3_exec(_handler, sql.c_str(), cb, arg, nullptr);
if (res != SQLITE_OK)
{
lg.LogMessage(error, "%s语句执行失败！失败原因：%s\n", sql.c_str(), sqlite3_errmsg(_handler));
return false;
}
return true;
}
void close()
{
sqlite3_close_v2(_handler);
}
private:
string _dbfile; // 数据库文件
sqlite3 *_handler; // 句柄
};
// 字符串操作类
class StrHelper
{
public:
/*
str: 输入型参数，目标要切分的字符串
res: 输出型，保存切分完毕的结果
sep: 指定的分割符
*/
static ssize_t SpiltStr(const string &str, vector<string> &res, string &sep)
{
// 采用boost库进行切割
// 传入参数：切割后传给谁，切割源文件，分割符（is_any_of说明遇到这个分割符就切割），是否压缩（on表示压缩，即遇到多个分割符合并为1个；off表示不压缩，不合并）
boost::split(res, str, boost::is_any_of(sep), boost::algorithm::token_compress_on);
return res.size(); // 返回切割数量
}
};
// UUID生成器类
class UUIDHelper
{
public:
// 获取通用唯一识别码，此处采用生成8个随机数字，加上8字节序号共16字节数组生成32位16进制字符的组合形式
static string Getuuid()
{
stringstream ss; // 处理字符串流，使读取字符串中的数据与输入流方式相似
// 1、生成8个0-255之间的随机数，用十六进制表示，共2*8=16位
random_device rd; // 随机数设备对象，通常用来获取非确定性的随机种子
// ssize_t RDnum = rd(); // 如果直接利用其重载的()获取随机数，获得的是机器随机数，获取效率较低
// 此处采用只取一个真正的机器随机数，然后利用梅森旋转算法，获得一个伪随机数的方案
mt19937_64 gernator(rd()); // 伪随机数生成器
uniform_int_distribution<int> distribution(0, 255); // 限定随机数生成区间在0-255之间（对应十六进制的0x00到0xFF）
for (int i = 0; i < 8; i++)
{
ss << setw(2) /* 设置下一个输出项的宽度，即输出的字符至少占2列 */
<< setfill('0') /* 设置当使用setw进行宽度设定时，所使用的填充字符是0 */
<< hex /* 把数设置为十六进制展示 */ << distribution(gernator) /*生成随机数*/;
if (i == 3 || i == 5 || i == 7)
{
ss << "-";
}
}
// 2、生成序列号，ssize_t类型8个字节，用十六进制表示，共2*8=16位（依次递增）
static atomic<ssize_t> number(1); // 线程间的序列号生成，适合使用原子的变量，避免手动加锁解锁，以1开始；同时采用static使其生命周期伴随整个程序，只初始化一次
ssize_t num = number.fetch_add(1); // 递增
for (int i = 7; i >= 0; i--)
{
ss << setw(2) << setfill('0') << hex
<< ((num >> (i * 8) /*每个字节是8位，每次右移说明增加一个字节*/) & 0xff /*只保留处理的当前字节*/);
if (i == 6)
{
ss << "-";
}
}
return ss.str();
}
};
// 文件基础操作类
class FileHelper
{
public:
FileHelper(const string &filename) : _filename(filename)
{
}
// 文件是否存在
bool Exist()
{
struct stat st;
// 调用stat函数并检查返回值，可以判断指定路径的文件或目录是否存在
return (stat(_filename.c_str(), &st /* 用于接收关于文件状态的信息。如果函数调用成功，文件的状态信息会被填充到这个结构体中 */) == 0);
}
// 文件大小获取
size_t size()
{
struct stat st;
int res = stat(_filename.c_str(), &st);
if (res == -1) // 文件不存在
{
return 0;
}
return st.st_size;
}
// 文件读取
bool Read(char *buffer, size_t StartSet, size_t len)
{
// 1、打开文件
ifstream ifs(_filename,
ios::binary /* 以二进制模式打开文件；在二进制模式下，文件会被原样读取，不做任何文本模式的转换（如行结束符的转换），这对读取非文本文件尤为重要 */
| ios::in /* 表明文件流用于输入操作，即从文件中读取数据 */);
if (!ifs.is_open())
{
lg.LogMessage(error, "%s文件打开失败！\n", _filename.c_str());
return false;
}
// 2、跳转至文件读取位置
// seekg函数用于设置读取文件的起始位置，ios::beg:是代表文件的起始位置位置标记，与StartSet结合使用时，表明从文件的开头向后偏移StartSet个字节来设置新的读取位置。
ifs.seekg(StartSet, ios::beg);
// 3、读取文件数据
ifs.read(buffer, len);
if (!ifs.good()) // 检查是否成功读写
{
lg.LogMessage(error, "%s文件读取数据失败！\n", _filename.c_str());
ifs.close();
return false;
}
// 4、关闭文件流
ifs.close();
return true;
}
bool Read(string &buffer)
{
buffer.resize(this->size());
return Read(&buffer[0], 0, this->size());
}
// 文件写入
bool Write(const char *buffer, size_t StartSet, size_t len)
{
// 1、打开文件（这里需要打开读取，后面又要写入，故需要使用fstream
fstream fs(_filename, ios::binary | ios::in | ios::out);
if (!fs.is_open())
{
lg.LogMessage(error, "%s文件打开失败！\n", _filename.c_str());
return false;
}
// 2、跳转至写入位置
fs.seekg(StartSet, ios::beg);
// 3、写入文件数据
fs.write(buffer, len);
if (!fs.good())
{
lg.LogMessage(error, "%s文件写入数据失败！\n", _filename.c_str());
fs.close();
return false;
}
// 4、关闭文件流
fs.close();
return true;
}
bool Write(string &buffer)
{
return Write(buffer.c_str(), 0, buffer.size());
}
// 文件创建/删除（采用静态函数提高效率）
static bool CreateFile(const string &filename)
{
fstream fs(filename, ios::binary | ios::out); // ios::out模式下，如果指定文件不存在，会创建该文件
if (!fs.is_open())
{
lg.LogMessage(error, "创建%s文件失败！\n", filename.c_str());
return false;
}
fs.close();
return true;
}
static bool DeleteFile(const string &filename)
{
return (remove(filename.c_str()) == 0);
}
// 目录创建/删除（采用静态函数提高效率）
static bool CreateDir(const string &path)
{
// 创建目录时，路径如果是多重路径，需要从一开始开始创建
int pos = 0;
int StartSet = 0;
while (pos < path.size())
{
pos = path.find("/", StartSet);
if (pos == string::npos)
{
return (mkdir(path.c_str(), 0775) == 0);
}
string subpath = path.substr(0, pos); // 每一层目录
int res = mkdir(subpath.c_str(), 0775);
if (res != 0 && errno != EEXIST) // 目录创建失败且不是因为目录已存在
{
lg.LogMessage(error, "创建%s目录失败！失败原因：%s", subpath.c_str(), strerror(errno));
return false;
}
StartSet = pos + 1;
}
return true;
}
static bool DeleteDir(const string &path)
{
string cmd = "rm -rf " + path;
return (system(cmd.c_str()) != -1); // 使用system函数执行构造的shell命令
}
// 传入文件名，获取其父目录路径，采用静态函数，减少调用，提高效率
static string GetParentDir(const string &path)
{
int pos = path.find_last_of("/"); // 从后往前找/分隔符
if (pos == string::npos)
{
return "./"; // 找不到，说明在根目录
}
return path.substr(0, pos);
}
// 修改文件名
bool Rename(const string &Newname)
{
return (rename(_filename.c_str(), Newname.c_str()) == 0);
}
private:
string _filename; // 文件名
};
}
2、线程池
生产者在发布消息成功后，需要通过信道通知消费者来消费消息，而这个操作需要网络传输，如果在主进程中进行，会阻塞当前操作，降低效率；因此最好实现一个线程池，把这个任务放在线程池中，交给线程池来实现。
2.1 std::future类
std::future是C++11标准库中的一个模板类，用来表示异步操作的结果。当我们在多线程编程中使用异步任务（异步任务执行时，可以在等待其结果的同时执行其它任务，可以稍后获取其结果，不会一直阻塞；就算只有一个主线程也可以体现这种异步）时，std::future可以帮助我们在需要的时候获取任务的执行结果。
std::future的⼀个重要特性是可以阻塞当前线程直到异步操作的完成，进而确保我们在获取结果时不会遇到未完成的操作。
std::future类的应用场景
1、获取异步任务的结果：当后台需要执行一些耗时操作（如网络请求等），为防止主线程因为此操作而阻塞，我们可以把该操作放在其他线程中执行，该线程称作异步工作线程。通过将这些任务和主线程分离，我们可以实现多任务并行处理，有效提高效率。
2、并发控制：在多线程编程中，我们可能需要等待某些任务完成后才能继续执行其他操作，通过使用std::future，我们可以实现线程的同步，确保在任务完成、获取结果后再继续进行其他操作。
3、更安全的方式获取异步任务的结果：可以使用std::future::get()函数来获取任务的结果，此函数会阻塞当前线程，直到异步任务完成。这样可以确保我们调用std::future::get()一定可以获得异步任务的结果。
2.2 std::packaged_task模板
std::packaged_task是一个模板类，实例化出的对象可以对一个函数进行二次封装，即把任务封装保存起来；可以通过get_future()函数获取一个future对象来得到这个异步任务的执行结果
常规情况下，std::packaged_task的模板参数是函数的各个参数与返回值。但是也可以接受一个可调用对象，然后只给一个返回值，std::packaged_task的构造函数根据会自动推导参数
示例代码如下：
#include <iostream>
#include <future>
#include <chrono>
// 实际执行的任务函数
int add(int x, int y)
{
std::this_thread::sleep_for(std::chrono::seconds(2)); // 模拟耗时操作
return x + y;
}
int main()
{
// 1、封装任务（由于封装的是一个函数，所以模板参数是函数的各个参数）
std::packaged_task<int(int, int)> task(add); // task类可以作为一个可调用对象调用函数执行任务，但其本身不能当做函数使用（不可以作为线程的入口函数）
// 2、获取其关联的future对象
std::future<int> result_future = task.get_future();
// 3、让任务执行（此处必须执行，否则后面future对象get()获取任务结果的时候会阻塞当前进程）
task(1, 2);
// 4、继续执行其他代码
std::cout << "异步任务执行期间，继续执行其他任务..." << std::endl;
// 5、等待一段时间，确保task已经完成
std::this_thread::sleep_for(std::chrono::seconds(3));
// 6、获取结果
int res = result_future.get();
std::cout << "任务执行结果为：" << res << std::endl;
return 0;
}
运行结果（先看见第一行输出，过了2秒才看见第二行）
2.3 线程池的实现
由于本项目中给予线程池执行的任务是固定的，因此本项目采用C++11标准库的模板类future、加上packaged_taskd的组合来实现
#pragma once
#include <iostream>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>
#include <functional>
using namespace std;
namespace ns_ThreadPool
{
class ThreadPool
{
public:
using ptr = shared_ptr<ThreadPool>;
using task_func = function<void(void)>; // 定义函数对象类型，该函数只是package_task封装的lambda表达式匿名函数
ThreadPool(int thread_count = 1) : _stop(false) // 传入线程数量
{
for (int i = 0; i < thread_count; i++)
{
// emplace_back会直接在thread_pool结尾构造一个thread线程对象，减少拷贝，然后该对象调用entry函数
// 在 C++ 中，你可以直接使用指向类成员函数的指针来构造线程，需要提供一个指向该类实例的指针，以便线程知道应该在哪个对象上调用成员函数。
_threads.emplace_back(&ThreadPool::entry, this);
}
}
~ThreadPool()
{
stop();
}
// 可变参数模板声明，第一个是F是一个类型参数，代表一个可调用对象（如函数、函数指针、lambda表达式或者函数对象等）。typename ...Args是可变参数
template <typename F, typename... Args>
// 传入参数：首先传入一个函数，然后传入函数所需的参数（即要处理的数据），采用万能引用，既能接收左值又能接收右值
// 该函数会把传入的参数封装成异步任务（packages_task），使用lambda表达式生成匿名函数对象放进任务池中，由工作线程取出执行
// auto xxx() -> returntype 函数返回类型推导
// decltype(func(args...))是C++中的一种类型推导表达式，将变量类型声明为表达式执行的类型，push函数的返回类型将根据你传递给它的函数func及参数调用的结果类型自动确定
// future<decltype(func(args...))> 是C++中的一种类型声明，用于定义一个 std::future 对象，该对象的类型与函数 func 被调用时的返回类型相匹配。
auto push(F &&func, Args &&.../*定义可变参数包时，把...放在参数类型的后面*/ args) -> future<decltype(func(args...))> // 返回的是future类型
{
// 1、将传入的函数封装成packages_task任务
// ...放在arg的后面是对之前已经定义好的可变参数包的展开操作，当args...这样的表达式出现在函数调用、类型推导或其他需要具体参数列表的地方时，...指示编译器将之前收集的所有参数（即Args中的每个参数）逐一插入到该位置
using return_type = decltype(func(args...)); // 把func函数的返回值定一个别名
auto tmp_func = bind(forward<F>(func), forward<Args>(args)...); // 创建一个可调用对象，采用完美转发让参数的左右值属性不变
auto task = make_shared<packaged_task<return_type()>>(tmp_func); // packaged_task可以接受一个可调用对象。构造函数能够从可调用对象中推断出参数类型。
future<return_type> fu = task->get_future();
// 2. 构造一个lambda匿名函数对象（捕获任务对象），传递至任务池中执行任务
// 限定作用域，确保在执行完添加任务到任务池的操作后立即解锁，减少锁的持有时间，提高程序的并发性能
{
unique_lock<mutex> lock(_mutex); // 管理和自动释放互斥锁`
// 插入一个lambda表达式，参数为空，里面的函数体也没有返回值，故是一个void(void)函数对象，符合条件
_taskpool.push_back([task]()
{ (*task)(); });
_cv.notify_one(); // 任务池中有新的任务加入，唤醒一个等待中的工作线程，尝试获取任务并执行
}
return fu;
}
// 停止运行
void stop()
{
if (_stop == true)
{
return;
}
_stop = true;
_cv.notify_all(); // 唤醒所有阻塞进程
for (auto &thread : _threads)
{
thread.join(); // 回收清理线程资源
}
}
private:
// 入口函数，内部从任务池中取出任务进行执行
void entry()
{
while (!_stop)
{
vector<task_func> tmp_taskpool;
{
unique_lock<mutex> lock(_mutex); // 定义一个锁
// 条件变量等待，传入参数：互斥锁，工作线程被唤醒的条件（用函数）
_cv.wait(lock, [this]()
{ return _stop || !_taskpool.empty(); }); // 当_stop为真（被置位了，即停止工作了）或任务池中有任务（即不为空）时，谓词返回true，wait函数将返回，线程将继续执行
tmp_taskpool.swap(_taskpool); // 把任务一次性全部交给任务池进行执行，省去一次次拿任务
}
for (auto &task_func : tmp_taskpool)
{
task_func();
}
}
}
atomic<bool> _stop; // 线程池是否为停止状态，是个原子对象，从不同的线程访问这个包含的值不会导致数据竞争，操作不可被打断
vector<task_func> _taskpool; // 函数任务池，因为packages_task不能作为线程的入口函数交给工作线程使用，所以每一个package_task都会封装成一个lambda表达式匿名函数放入该任务池
mutex _mutex; // 锁，用来保障互斥
condition_variable _cv; // C++11线程库中的条件变量，用来保证同步
// 互斥锁和条件变量一般都定义在线程池之前，防止线程池使用的时候锁和信号量没有定义
vector<thread> _threads; // 线程池
};
}
六、模块划分与具体实现
项目模块关系图
1、服务端模块
1.1 持久化数据管理中心模块
该模块分为消息数据管理、交换机数据管理、队列数据管理、绑定数据管理四个模块，分别实现持久化存储以及数据的增删查。
1.1.1 交换机数据管理模块
要管理的主要数据（描述一个交换机应该有什么数据）
1、交换机名称：交换机的唯一标识
2、交换机类型：分为直接交换、广播交换、主题交换；决定了消息的转发方式（定制在了protobuf协议中）
// 交换机类型的枚举
enum ExchangeType
{
UNKNOWNTYPE = 0; // 由于枚举必须从0开始，所以设置一下，没有实际意义
direct = 1; // 直接交换
fanout = 2; // 广播交换
topic = 3; // 主题交换
}
3、持久化标志：决定当前交换机信息是否需要持久化存储
4、交换机的其他一些参数
对交换机的管理操作
1、创建 / 声明交换机：有强断言思想存在，如果交换机存在就不再创建，没有再创
2、删除交换机（*注意：每个交换机都会绑定一个 / 多个队列，因此删除交换机之前，必须删除其相关的绑定消息*）
3、获取指定名称的交换机
4、获取当前交换机的数量
// 交换机有关操作
// 用户发布消息时，先把消息发给交换机，再由交换机发给指定队列
#pragma once
#include "../MQcommon/Helper.hpp"
#include "../MQcommon/Log.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include <google/protobuf/map.h>
#include <iostream>
#include <sstream>
#include <string>
#include <mutex>
#include <memory>
#include <unordered_map>
using namespace std;
using namespace ns_helper;
namespace ns_exchange
{
// 交换机类
class Exchange
{
public:
using ptr = shared_ptr<Exchange>; // 用只能指针管理交换机类
Exchange()
{
}
Exchange(const string &name, MQ_message::ExchangeType type, bool durable, bool auto_delete, const google::protobuf::Map<std::string, std::string> &args) : _name(name), _type(type), _durable(durable), _auto_delete(auto_delete), _args(args)
{
// debug
std::cout << "testing>exchange: " << _name << std::endl;
}
// args存储键值对，会以key=val & key=val……的形式存储在数据库中，因此需要从数据库内部解析str_args字符串，再把内容存储到成员中
void SetArgs(const string &str_args)
{
vector<string> sub_args;
string sep = "&";
StrHelper::SpiltStr(str_args, sub_args, sep);
for (auto &str : sub_args)
{
int pos = str.find("=");
string key = str.substr(0, pos);
string val = str.substr(pos + 1);
_args[key] = val;
}
}
// 将args中的内容进行序列化后，返回一个字符串
string GetArgs()
{
string res;
if (_args.empty())
return res;
for (auto i = _args.begin(); i != _args.end(); i++)
{
res += (i->first + "=" + i->second + "&");
}
res.pop_back();
return res;
}
string _name; // 交换机名称
MQ_message::ExchangeType _type; // 交换机类型
bool _durable; // 交换机是否持久化标志
bool _auto_delete; // 是否自动删除标志
// unordered_map<string, string> _args; // 不符合谷歌规范
google::protobuf::Map<string, string> _args; // 其他参数
};
// 交换机数据持久化管理类（内部的交换机数据管理类）——数据存储在sqlite数据库中，提供交换机的增删查以及数据库建删表服务，
class ExchangeMapper
{
public:
// 传入数据库名称，初始化
ExchangeMapper(const string &dbfile) : _sql_helper(dbfile)
{
string path = FileHelper::GetParentDir(dbfile);
FileHelper::CreateDir(path);
assert(_sql_helper.open());
CreateTable();
}
// 创建/删除交换机数据表
void CreateTable()
{
string CreateTableSQL = "create table if not exists ExchangeTable(name varchar(32) primary key, type int, durable int, auto_delete int, args varchar(128));";
bool res = _sql_helper.exec(CreateTableSQL, nullptr, nullptr);
if (res == false)
{
lg.LogMessage(error, "交换机数据库表建立失败!\n");
abort(); // 立即终止程序执行
}
lg.LogMessage(info, "交换机数据库表建立成功！\n");
}
void DeleteTable()
{
string DeleteTableSQL = "drop table if exists ExchangeTable";
bool res = _sql_helper.exec(DeleteTableSQL, nullptr, nullptr);
if (res == false)
{
lg.LogMessage(error, "交换机数据库表删除失败！\n");
abort();
}
}
// 新增交换机
bool Insert(Exchange::ptr &exp)
{
stringstream ss;
ss << "insert into ExchangeTable values(";
ss << "'" << exp->_name << "', ";
ss << exp->_type << ", ";
ss << exp->_durable << ", ";
ss << exp->_auto_delete << ", ";
ss << "'" << exp->GetArgs() << "');";
return _sql_helper.exec(ss.str(), nullptr, nullptr);
}
// 移除交换机，传入交换机名称移除交换机
bool Remove(const string &name)
{
stringstream ss;
ss << "delete from ExchangeTable where name=";
ss << "'" << name << "';";
return _sql_helper.exec(ss.str(), nullptr, nullptr);
}
// 查询注意：用哈希表管理交换机名称和交换机对象之间的映射关系，通过交换机名称查找交换机对象，提高效率
// 恢复历史数据：查询所有的交换机数据（此处采用unordermap对象，直接建立起交换机名称和交换机对象之间的映射关系，简化操作）
unordered_map<string, Exchange::ptr> Recovery()
{
string sql = "select name, type, durable, auto_delete, args from ExchangeTable";
unordered_map<string, Exchange::ptr> result;
_sql_helper.exec(sql, SelectCallback, &result);
return result;
}
private:
// 回调函数
// 传入参数：参数，列数，行数（指向一行查询结果的二维指针，每一项是一个指向该行某一列数据的字符指针），字段名称（用处不大）
// row参数是由SQLite库内部提供的，它代表了查询结果集中当前行的每一列数据
static int SelectCallback(void *args, int numcol, char **row, char **fields)
{
// 使用 void* 指针之前，通常需要通过强制类型转换将其转换为具体的类型指针
unordered_map<string, Exchange::ptr> *res = (unordered_map<string, Exchange::ptr> *)args;
auto exp = make_shared<Exchange>(); // 需要无参构造函数提供支持
exp->_name = row[0];
exp->_type = (MQ_message::ExchangeType)(stoi(row[1]));
exp->_durable = (bool)stoi(row[2]);
exp->_auto_delete = (bool)stoi(row[3]);
if (row[4]) // 有可能不存在其他参数，为空
{
exp->SetArgs(row[4]);
}
res->insert(make_pair(exp->_name, exp));
return 0;
}
SQLiteHelper _sql_helper; // 操纵SQLite数据库对象
};
// 交换机数据内存管理类（真正对外的交换机数据管理类），可能会在多个线程中使用，要确保线程安全
class ExchangeManager
{
public:
using ptr = shared_ptr<ExchangeManager>;
ExchangeManager(const string &dbfile) : _mapper(dbfile)
{
_exchanges = _mapper.Recovery();
}
// 声明/添加交换机
bool DeclareExchange(const string &name, MQ_message::ExchangeType type, bool durable, bool auto_delete, const google::protobuf::Map<string, string> &args)
{
unique_lock<mutex> lock(_mutex); // 加锁
auto pos = _exchanges.find(name);
if (pos != _exchanges.end()) // 说明交换机已存在
{
// lg.LogMessage(warning, "%s交换机已存在！\n", name.c_str());
return true;
}
auto Newexchange = make_shared<Exchange>(name, type, durable, auto_delete, args);
if (durable)
{
bool res = _mapper.Insert(Newexchange);
if (res == false)
{
return false;
}
}
_exchanges[name] = Newexchange;
lg.LogMessage(info, "%s 交换机创建成功！\n", name.c_str());
return true;
}
// 删除交换机
bool DeleteExchange(const string &name)
{
unique_lock<mutex> lock(_mutex); // 加锁
auto pos = _exchanges.find(name);
if (pos == _exchanges.end())
{
lg.LogMessage(warning, "%s交换机不存在！\n", name.c_str());
return false;
}
if (pos->second->_durable)
{
bool res = _mapper.Remove(name);
if (res == false)
{
return false;
}
}
_exchanges.erase(name);
return true;
}
// 获取交换机数目
int Size()
{
return _exchanges.size();
}
// 判定某交换机是否存在
bool ExistExchange(const string &name)
{
unique_lock<mutex> lock(_mutex);
auto pos = _exchanges.find(name);
if (pos == _exchanges.end())
{
return false;
}
return true;
}
// 获取指定交换机对象
Exchange::ptr SelectExchange(const string &name)
{
unique_lock<mutex> lock(_mutex);
auto it = _exchanges.find(name);
if (it == _exchanges.end())
{
return Exchange::ptr();
}
return it->second;
}
// 清理所有交换机（供测试用，外部用户用处不大）
void ClearAllExchanges()
{
unique_lock<mutex> lock(_mutex);
_exchanges.clear();
_mapper.DeleteTable();
}
private:
ExchangeMapper _mapper; // 数据持久化类
unordered_map<string, Exchange::ptr> _exchanges; // 交换机名称和交换机对象之间的映射关系，便于查询
mutex _mutex; // 互斥锁，确保线程安全，防止冲突
};
}
1.1.2 队列数据管理模块
队列要管理的主要数据
1、队列名称：队列唯一标识
2、持久化存储标志：决定了是否要把队列的信息持久化存储起来，即重启之后，这个队列是否还存在。
3、是否独占标志：即同一时间内是否只能有一个线程可以访问该队列
4、队列的其他一些参数
对队列的管理操作
1、创建队列
2、删除队列
3、获取指定队列消息
4、获取队列数量
5、获取所有队列名称
// 队列数据管理类，大体与交换机类类似
#pragma once
#include "../MQcommon/Helper.hpp"
#include "../MQcommon/Log.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include <iostream>
#include <sstream>
#include <string>
#include <mutex>
#include <memory>
#include <unordered_map>
using namespace std;
using namespace ns_helper;
namespace ns_queue
{
class MsgQueue
{
public:
using Queue_ptr = shared_ptr<MsgQueue>;
MsgQueue()
{
}
MsgQueue(const string &name, bool durable, bool exclusive, bool auto_delete, const google::protobuf::Map<std::string, std::string> &args) : _name(name), _durable(durable), _exclusive(exclusive), _auto_delete(auto_delete), _args(args)
{
}
// 参数切，即反序列化
void SetArgs(const string &str_args)
{
vector<string> sub_args;
string sep = "&";
StrHelper::SpiltStr(str_args, sub_args, sep);
for (auto &str : sub_args)
{
int pos = str.find("=");
string key = str.substr(0, pos);
string val = str.substr(pos + 1);
_args[key] = val;
}
}
// 将args中的内容进行序列化后，返回一个字符串
string GetArgs()
{
string res;
if (_args.empty())
return res;
for (auto i = _args.begin(); i != _args.end(); i++)
{
res += (i->first + "=" + i->second + "&");
}
res.pop_back();
return res;
}
string _name; // 队列名称
bool _durable; // 队列是否持久化标志
bool _exclusive; // 是否独占标志（同一时间内只能有一个线程可以访问该队列）
bool _auto_delete; // 是否自动删除标志
google::protobuf::Map<string, string> _args; // 其他参数
};
class MsgQueueMapper
{
public:
MsgQueueMapper(const string &dbfile) : _sql_helper(dbfile)
{
string path = FileHelper::GetParentDir(dbfile);
FileHelper::CreateDir(path);
assert(_sql_helper.open());
CreateTable();
}
// 创建/删除队列数据表
void CreateTable()
{
string CreateTableSQL = "create table if not exists QueueTable(name varchar(32) primary key, durable int, exclusive int, auto_delete int, args varchar(128));";
bool res = _sql_helper.exec(CreateTableSQL, nullptr, nullptr);
if (res == false)
{
lg.LogMessage(error, "队列数据库表建立失败!\n");
abort(); // 立即终止程序执行
}
lg.LogMessage(info, "队列数据库表建立成功！\n");
}
void DeleteTable()
{
string DeleteTableSQL = "drop table if exists QueueTable";
bool res = _sql_helper.exec(DeleteTableSQL, nullptr, nullptr);
if (res == false)
{
lg.LogMessage(error, "队列数据库表删除失败！\n");
abort();
}
}
// 新增队列数据
bool Insert(MsgQueue::Queue_ptr &exp)
{
stringstream ss;
ss << "insert into QueueTable values(";
ss << "'" << exp->_name << "', ";
ss << exp->_durable << ", ";
ss << exp->_exclusive << ", ";
ss << exp->_auto_delete << ", ";
ss << "'" << exp->GetArgs() << "');";
return _sql_helper.exec(ss.str(), nullptr, nullptr);
}
// 删除队列数据
bool Remove(const string &name)
{
stringstream ss;
ss << "delete from QueueTable where name=";
ss << "'" << name << "';";
return _sql_helper.exec(ss.str(), nullptr, nullptr);
}
// 查询所有队列数据
unordered_map<string, MsgQueue::Queue_ptr> GetAll()
{
string sql = "select* from QueueTable";
unordered_map<string, MsgQueue::Queue_ptr> result;
_sql_helper.exec(sql, SelectCallback, &result);
return result;
}
private:
static int SelectCallback(void *args, int numcol, char **row, char **fields)
{
// 使用 void* 指针之前，通常需要通过强制类型转换将其转换为具体的类型指针
unordered_map<string, MsgQueue::Queue_ptr> *res = (unordered_map<string, MsgQueue::Queue_ptr> *)args;
auto exp = make_shared<MsgQueue>(); // 需要无参构造函数提供支持
exp->_name = row[0];
exp->_durable = (bool)stoi(row[1]);
exp->_exclusive = (bool)stoi(row[2]);
exp->_auto_delete = (bool)stoi(row[3]);
if (row[4]) // 有可能不存在其他参数，为空
{
exp->SetArgs(row[4]);
}
res->insert(make_pair(exp->_name, exp));
return 0;
}
SQLiteHelper _sql_helper;
};
class MsgQueueManager
{
public:
using ptr = shared_ptr<MsgQueueManager>;
MsgQueueManager(const string &dbfile) : _mapper(dbfile)
{
_queues = _mapper.GetAll();
}
// 声明/添加队列
bool DeclareQueue(const string &name, bool durable, bool exclusive, bool auto_delete, const google::protobuf::Map<string, string> &args)
{
unique_lock<mutex> lock(_mutex); // 加锁
auto pos = _queues.find(name);
if (pos != _queues.end()) // 说明交换机已存在
{
// lg.LogMessage(warning, "%s队列已存在！\n", name.c_str());
return true;
}
auto Newqueue = make_shared<MsgQueue>(name, durable, exclusive, auto_delete, args);
if (durable)
{
bool res = _mapper.Insert(Newqueue);
if (res == false)
{
return false;
}
}
_queues[name] = Newqueue;
return true;
}
// 删除队列
bool DeleteQueue(const string &name)
{
unique_lock<mutex> lock(_mutex); // 加锁
auto pos = _queues.find(name);
if (pos == _queues.end())
{
lg.LogMessage(warning, "%s队列不存在！\n", name.c_str());
return false;
}
if (pos->second->_durable)
{
bool res = _mapper.Remove(name);
if (res == false)
{
return false;
}
}
_queues.erase(name);
return true;
}
// 获取指定交换机对象
MsgQueue::Queue_ptr SelectQueue(const string &name)
{
unique_lock<mutex> lock(_mutex);
auto pos = _queues.find(name);
if (pos == _queues.end())
{
lg.LogMessage(warning, "%s队列不存在，无法获取！\n", name.c_str());
return MsgQueue::Queue_ptr();
}
return pos->second;
}
// 获取队列数目
int Size()
{
return _queues.size();
}
// 获取所有消息队列，用于从数据库恢复队列历史信息
unordered_map<string, MsgQueue::Queue_ptr> GetAllQueues()
{
unique_lock<mutex> lock(_mutex);
return _queues;
}
// 判定某队列是否存在
bool ExistQueue(const string &name)
{
unique_lock<mutex> lock(_mutex);
auto pos = _queues.find(name);
if (pos == _queues.end())
{
return false;
}
return true;
}
// 清理所有交换机（供测试用，外部用户用处不大）
void ClearAllQueues()
{
unique_lock<mutex> lock(_mutex);
_queues.clear();
_mapper.DeleteTable();
}
private:
MsgQueueMapper _mapper; // 数据持久化类
unordered_map<string, MsgQueue::Queue_ptr> _queues; // 队列名称和队列对象之间的映射关系，便于查询
mutex _mutex; // 互斥锁，确保线程安全，防止冲突
};
}
1.1.3 绑定数据管理模块
在该模块中，使用SQLiteHelper类建立一个表，里面有三个属性——交换机名称、队列名称、以及交换队列的绑定规则binding_key，这个表记载着所有的绑定信息数据，在操纵绑定信息的时候，就是使用SQL指令操纵这张表。
恢复或重新建立交换机与消息队列之间的绑定关系信息时，也是根据从数据库查询得到的每一条绑定记录，逐步构建或更新ExchangesBindingMap中的数据结构
管理的主要数据
1、交换机名称
2、队列名称
3、binding_key（绑定秘钥）：描述了主题交换 / 直接交换类型交换机的消息发布匹配规则，由数字、字符、_ 、# 、. 、* 组成（*eg : news.music.#*）
管理的操作
1、添加绑定
2、解除绑定
3、获取指定交换机相关的所有绑定信息（当消息发布到交换机，交换机通过这些绑定信息将消息发布到指定队列；删除交换机时，也要删除相关的绑定信息）*
4、获取指定队列相关的所有绑定信息
5、获取绑定信息数量
// 绑定信息类，本质上就是描述一个交换机上关联了哪些队列
#pragma once
#include "../MQcommon/Helper.hpp"
#include "../MQcommon/Log.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include <iostream>
#include <sstream>
#include <string>
#include <mutex>
#include <memory>
#include <unordered_map>
using namespace std;
using namespace ns_helper;
namespace ns_binding
{
// 绑定信息类
class Binding
{
public:
using ptr = shared_ptr<Binding>;
Binding()
{
}
Binding(string exchange_name, string msgqueue_name, string binding_key) : _exchange_name(exchange_name), _msgqueue_name(msgqueue_name), _binding_key(binding_key)
{
}
string _exchange_name; // 交换机名称
string _msgqueue_name; // 队列名称
string _binding_key; // 分发匹配规则，决定哪些数据可以被从交换机放入队列
};
// 队列与绑定信息一一对应，相当于给某个交换机上绑定一个队列，因此一个交换机上可以绑定多个队列，即有多个队列的绑定信息，便于操作
// 队列集合：队列名与绑定信息映射关系的集合，便于通过队列名查找绑定信息
using MsgQueuesBindingMap = unordered_map<string, Binding::ptr>;
// 交换机集合：交换机名称与绑定其之上的队列集合的映射关系的集合，包含了所有的绑定信息，以交换机为单元进行区分
using ExchangesBindingMap = unordered_map<string, MsgQueuesBindingMap>;
// 绑定信息数据持久化类
class BindingMapper
{
public:
BindingMapper(const string &dbfile) : _sql_helper(dbfile)
{
string path = FileHelper::GetParentDir(dbfile);
FileHelper::CreateDir(path);
assert(_sql_helper.open());
CreateTable();
}
// 创建/删除绑定信息数据表
void CreateTable()
{
string CreateSql = "create table if not exists BindingTable (exchange_name varchar(32), msgqueue_name varchar(32), binding_key varchar(128));";
_sql_helper.exec(CreateSql, nullptr, nullptr);
}
void DeleteTable()
{
string DeleteSql = "drop table BindingTable;";
_sql_helper.exec(DeleteSql, nullptr, nullptr);
}
// 新增绑定信息数据
bool Insert(Binding::ptr &binding)
{
stringstream sql;
sql << "insert into BindingTable values(";
sql << "'" << binding->_exchange_name << "', ";
sql << "'" << binding->_msgqueue_name << "', ";
sql << "'" << binding->_binding_key << "');";
return _sql_helper.exec(sql.str(), nullptr, nullptr);
}
// 移除指定绑定信息数据
void Remove(const string &Ename, const string &Qname)
{
stringstream sql;
sql << "delete from BindingTable where ";
sql << "exchange_name='" << Ename << "' and ";
sql << "msgqueue_name='" << Qname << "';";
_sql_helper.exec(sql.str(), nullptr, nullptr);
}
// 移除指定交换机相关的绑定信息数据
void RemoveExchangeBinding(const string &Ename)
{
stringstream sql;
sql << "delete from BindingTable where ";
sql << "exchange_name='" << Ename << "';";
_sql_helper.exec(sql.str(), nullptr, nullptr);
}
// 移除指定队列的相关绑定信息数据
void RemoveMsgQueueBinding(const string &Qname)
{
stringstream sql;
sql << "delete from BindingTable where ";
sql << "msgqueue_name='" << Qname << "';";
_sql_helper.exec(sql.str(), nullptr, nullptr);
}
// 查询所有绑定信息数据（即所有的交换机集合），用于重启服务器时的历史数据恢复
ExchangesBindingMap Recovery()
{
string RecoverSql = "select exchange_name, msgqueue_name, binding_key from BindingTable;";
ExchangesBindingMap res;
_sql_helper.exec(RecoverSql, SelectCallback, &res);
return res;
}
private:
static int SelectCallback(void *args, int numcol, char **row, char **fields)
{
ExchangesBindingMap *result = (ExchangesBindingMap *)args;
Binding::ptr BindingPtr = make_shared<Binding>(row[0], row[1], row[2]);
// 根据从数据库查询得到的每一条绑定记录，逐步构建或更新ExchangesBindingMap中的数据结构，以恢复或重新建立交换机与消息队列之间的绑定关系信息
// 如果result中没有给定的交换机名称，会自动创建一个新的MsgQueuesBindingMap并插入到ExchangesBindingMap中，再把新的MsgQueuesBindingMap交给MQmap，这是一种便利的动态数据填充方式
// 为了防止有同名交换机绑定信息，不可以直接添加数据，防止绑定信息被覆盖
// 因此得先获取交换机对应的映射对象，往里边添加数据，若这时候没有交换机对应的映射信息，因此这里的获取要使用引用（会保证不存在则自动创建）
MsgQueuesBindingMap &MQmap = (*result)[BindingPtr->_exchange_name];
MQmap[BindingPtr->_msgqueue_name] = BindingPtr;
return 0;
}
SQLiteHelper _sql_helper;
};
// 绑定信息数据管理类
class BindingManager
{
public:
using ptr = shared_ptr<BindingManager>;
BindingManager(const string &dbfile) : _mapper(dbfile)
{
_bindings = _mapper.Recovery(); // 获取全部数据
}
// 创建绑定信息，并添加管理（存在添加，不存在创建）
bool Bind(const string &Ename, const string &Qname, const string &key, bool durable)
{
unique_lock<mutex> lock(_mutex);
auto it = _bindings.find(Ename);
if (it != _bindings.end() && it->second.find(Qname) != it->second.end())
{
// 绑定信息已经存在，不用创建
return true;
}
Binding::ptr bp = make_shared<Binding>(Ename, Qname, key);
// 绑定信息是否需要持久化取决于交换机数据和队列数据是否要持久化
if (durable)
{
bool res = _mapper.Insert(bp);
if (!res)
{
return false;
}
}
auto &mqbp = _bindings[Ename];
mqbp[Qname] = bp;
return true;
}
// 解除指定的绑定信息
bool UnBind(const string &Ename, const string &Qname)
{
unique_lock<mutex> lock(_mutex);
auto Epos = _bindings.find(Ename);
if (Epos == _bindings.end())
{
lg.LogMessage(warning, "交换机集合中没有%s交换机！\n", Ename.c_str());
return false;
}
_bindings.erase(Ename);
auto Qpos = _bindings[Ename].find(Qname);
if (Qpos == _bindings[Ename].end())
{
lg.LogMessage(warning, "%s交换机中不存在%s绑定队列！\n", Ename.c_str(), Qname.c_str());
return false;
}
_mapper.Remove(Ename, Qname);
_bindings[Ename].erase(Qname);
return true;
}
// 移除指定队列的所有绑定信息
void RemoveMsgQueueBindings(const string &Qname)
{
unique_lock<mutex> lock(_mutex);
_mapper.RemoveMsgQueueBinding(Qname);
for (auto i = _bindings.begin(); i != _bindings.end(); i++)
{
// 遍历每个交换机，移除每个交换机所绑定的该名称队列
i->second.erase(Qname);
}
}
// 移除某个交换机相关的所有绑定信息
void RemoveExchangeBinding(const string &Ename)
{
unique_lock<mutex> lock(_mutex);
auto Epos = _bindings.find(Ename);
if (Epos == _bindings.end())
{
lg.LogMessage(warning, "交换机集合中没有%s交换机！\n", Ename.c_str());
return;
}
_bindings.erase(Ename);
}
// 获取某一个交换机相关的所有绑定信息
MsgQueuesBindingMap GetExchangeBindings(const string &Ename)
{
unique_lock<mutex> lock(_mutex);
auto Epos = _bindings.find(Ename);
if (Epos == _bindings.end())
{
lg.LogMessage(warning, "交换机集合中没有%s交换机！\n", Ename.c_str());
return MsgQueuesBindingMap();
}
return _bindings[Ename];
}
// 获取指定交换机上的指定队列的绑定信息
Binding::ptr GetExchangeQueueBinding(const string &Ename, const string &Qname)
{
unique_lock<mutex> lock(_mutex);
auto Epos = _bindings.find(Ename);
if (Epos == _bindings.end())
{
lg.LogMessage(warning, "交换机集合中没有%s交换机！\n", Ename.c_str());
return Binding::ptr();
}
auto Qpos = _bindings[Ename].find(Qname);
if (Qpos == _bindings[Ename].end())
{
lg.LogMessage(warning, "%s交换机中不存在%s绑定队列！\n", Ename.c_str(), Qname.c_str());
return Binding::ptr();
}
return Qpos->second;
}
// 判断制定交换机上的指定队列的绑定信息是否存在
bool Exists(const string &Ename, const string &Qname)
{
unique_lock<mutex> lock(_mutex);
auto Epos = _bindings.find(Ename);
if (Epos == _bindings.end())
{
lg.LogMessage(warning, "交换机集合中没有%s交换机！\n", Ename.c_str());
return false;
}
auto Qpos = _bindings[Ename].find(Qname);
if (Qpos == _bindings[Ename].end())
{
lg.LogMessage(warning, "%s交换机中不存在%s绑定队列！\n", Ename.c_str(), Qname.c_str());
return false;
}
return true;
}
// 获取当前绑定信息队列数量
ssize_t Size()
{
unique_lock<mutex> lock(_mutex);
ssize_t sum = 0;
for (auto i = _bindings.begin(); i != _bindings.end(); i++)
{
// 遍历每个交换机，获得每个交换机上绑定队列的数目
sum += i->second.size();
}
return sum;
}
// 销毁所有绑定信息数据
void Clear()
{
unique_lock<mutex> lock(_mutex);
_mapper.DeleteTable();
_bindings.clear();
}
private:
mutex _mutex;
BindingMapper _mapper; // 持久化管理
ExchangesBindingMap _bindings; // 交换机集合
};
}
1.1.4 消息数据管理模块
1、消息信息
消息属性
ID：消息的唯一标识
持久化标志：是否对消息进行持久化
routing_key：消息发布到交换机后，与绑定队列的binding_key匹配，决定发布到哪个队列
消息主体：（消息的内容）
*（PS：以下是服务端为了管理而添加的信息）*
* *存储偏移量：当前消息对于文件起始位置的偏移量
消息长度：从偏移量位置取出该长度消息，解决黏包问题
是否有效：标志该消息是否属于有效消息，规定当有效消息比例低于总体消息的50%时，执行垃圾回收机制（存储的是字符0/1，保证修改后长度不变；需要存储到文件中，只够存储到文件中，才能在垃圾回收时判断是否需要回收）
** 上述信息，都被定制在了mq_msg.proto文件中，便于在网络中进行传输**
// 消息的投递模式
enum DeliveryMode
{
UNKNOWNMODE = 0;
undurable = 1; // 非持久化模式
durable = 2; // 持久化模式
}
// 消息属性的主体部分
message BasicProperties
{
string id = 1; // 消息ID
DeliveryMode delivery_mode = 2; // 消息的投递模式
string routing_key = 3; // 消息的routing_key，与Binding_key匹配
}
// 消息属性
message MQMessage
{
// 消息的有效载荷（都要持久化存储至文件中）
message Payload
{
BasicProperties properties = 1; // 消息的主体属性
string body = 2; // 消息的内容
string valid = 4; // 持久化消息是否有效（存储的是字符0/1，保证修改后长度不变；需要存储到文件中，只够存储到文件中，才能在垃圾回收时判断是否需要回收）
}
Payload payload = 1; // 消息的有效载荷
uint32 offset = 2; // 消息数据的相对位置（偏移量）
uint32 length = 3; // 消息的长度
}
2、消息管理
因为消息的一切操作都是以队列为单元的，所以我们要以队列为单元进行管理
管理的主要数据
1、持久化文件中有效消息的数量
2、持久化文件中总体消息的数量（通过与有效消息联合，可以计算出文件中有效消息的比例，以判断是否低于50%进行垃圾回收）
（PS：垃圾回收思路：把有效消息截取出来写在一个新文件中，然后删去源文件，把新文件的名字改成和源文件名字相同，这样文件中就仅剩有效消息。注意：在垃圾回收完毕之后，需要遍历一遍返回而来的有效消息链表，取出ID，到持久化消息的哈希表中寻找，如果出现没有找到的ID，说明持久化表更新不完全，需要把未更新的有效消息插进去）
3、消息链表：存有所有的待推送消息
4、持久化消息的哈希表，便于定位实际存储位置，在垃圾回收后更新消息数据
5、待确认消息的哈希表（内存中的数据）*（一条消息被推送给客户端，并不会立刻真正删除，而是会取出待推送链表加入待确认结构中，等到被确认后才会删除）*
管理操作
1、向队列新增消息
2、获取队首消息***（获取队首消息后，队首消息将从待推送消息链表中移除，成为待确认消息，加入待确认消息哈希表中）***
*** *** 3、确认消息：消息确认后，消息将从待确认消息表中移除，并根据消息的持久化模式，判断是否需要从持久化数据表中删去（为什么要从持久化数据表中删除？因为在在消息队列的设计中，消息通常只被消费一次，并且一旦成功处理后就不需要再次处理，故而删去）
4、恢复历史消息：获取文件中还保存着的持久化消息
1.2 虚拟机管理模块
虚拟机其实就是由交换机+队列+绑定+消息整合而来的整体逻辑单元，因此所谓的虚拟机管理模块，就是上述**四个模块的整合，整个项目主体都是以虚拟机为单元进行操作的 **
1.2.1 管理的主要数据
1、交换机数据管理句柄
2、队列数据管理句柄
3、绑定数据管理句柄
4、消息数据管理句柄
5、虚拟机名称*（本项目为简化操作，仅使用一个虚拟机，不实现多个虚拟机的管理操作）*
1.2.2 管理的操作
1、声明 / 删除交换机（*注意：每个交换机都会绑定一个 / 多个队列，因此删除交换机之前，必须删除其相关的绑定消息*）
2、声明 / 删除队列（*注意：删除队列之前，必须先删除与之相关的消息和消息数据*）
3、交换机和队列之间的绑定与解绑（*注意：绑定与解绑之前，必须确保有关交换机和队列存在*）
4、获取（消费）指定队列上的信息
5、确认指定队列上的指定消息
6、获取某个交换机上的所有绑定队列（*发布消息之前，获取交换机上的所有绑定队列，以判断这个消息需要发布到哪个队列中*）
7、发布消息：向某个消息队列新增消息
// 虚拟机头文件，整个项目主体都是以虚拟机为单元进行操作的，整合各个单元
// 本项目为简化操作，仅使用一个虚拟机，不实现多个虚拟机的管理操作
#pragma once
#include "Exchange.hpp"
#include "Queue.hpp"
#include "Binding.hpp"
#include "Message.hpp"
using namespace std;
using namespace ns_helper;
using namespace ns_exchange;
using namespace ns_queue;
using namespace ns_binding;
using namespace ns_message;
namespace ns_VirtualHost
{
class VirtualHost
{
public:
using ptr = shared_ptr<VirtualHost>;
VirtualHost(const string &Hname, const string &basedir, const string &dbfile) : _host_name(Hname),
_EXmp(make_shared<ExchangeManager>(dbfile)),
_MQmp(make_shared<MsgQueueManager>(dbfile)),
_BMmp(make_shared<BindingManager>(dbfile)),
_MMmp(make_shared<MessageManager>(basedir))
{
// 交换机、队列、交换机队列绑定信息在构造的时候都会从数据库中恢复历史消息；但是消息是在初始化消息队列的时候才恢复历史消息，因此需要遍历队列信息通过队列名称恢复
unordered_map<string, MsgQueue::Queue_ptr> queues = _MQmp->GetAllQueues();
for (auto &queue : queues)
{
_MMmp->InitQueueMessage(queue.first);
}
}
// 1、交换机的声明与删除
bool DeclareExchange(const string &Ename, MQ_message::ExchangeType type, bool durable, bool auto_delete, const google::protobuf::Map<string, string> &args)
{
return _EXmp->DeclareExchange(Ename, type, durable, auto_delete, args);
}
bool DeleteExchange(const string &Ename)
{
// 删除交换机的同时，与其相关的绑定信息也需要删除（不能删除绑定队列，绑定队列与交换机是两个独立的单元，只有绑定信息与交换机相关）
_BMmp->RemoveExchangeBinding(Ename);
return _EXmp->DeleteExchange(Ename);
}
// 判断交换机是否存在
bool ExistsExchange(const string &Ename)
{
return _EXmp->ExistExchange(Ename);
}
// 查询某个交换机
Exchange::ptr SelectExchange(const std::string &ename)
{
return _EXmp->SelectExchange(ename);
}
// 2、队列的声明与删除
bool DeclareQueue(const string &Qname, bool durable, bool exclusive, bool auto_delete, const google::protobuf::Map<string, string> &args)
{
// 初始化队列的消息句柄，进行消息的存储管理
_MMmp->InitQueueMessage(Qname);
// 创建队列
return _MQmp->DeclareQueue(Qname, durable, exclusive, auto_delete, args);
}
bool DeleteQueue(const string &Qname)
{
// 队列、队列消息和队列的绑定信息都要删除
_BMmp->RemoveMsgQueueBindings(Qname);
_MMmp->DestroyQueueMessage(Qname);
return _MQmp->DeleteQueue(Qname);
}
// 判断队列是否存在
bool ExistsQueue(const string &Qname)
{
return _MQmp->ExistQueue(Qname);
}
// 获取所有队列
unordered_map<string, MsgQueue::Queue_ptr> GetAllQueues()
{
return _MQmp->GetAllQueues();
}
// 3、交换机队列的绑定与解绑
bool ExchangeQueueBinding(const string &Ename, const string &Qname, const string &key)
{
Exchange::ptr exp = _EXmp->SelectExchange(Ename);
if (exp == nullptr)
{
lg.LogMessage(error, "交换机绑定队列失败，交换机不存在！\n");
return false;
}
MsgQueue::Queue_ptr qp = _MQmp->SelectQueue(Qname);
if (!_MQmp->ExistQueue(Qname))
{
lg.LogMessage(error, "交换机绑定队列失败，队列不存在\n");
return false;
}
// 当交换机和消息队列持久化模式都是持久化，绑定信息才持久化
return _BMmp->Bind(Ename, Qname, key, exp->_durable && qp->_durable);
}
bool ExchangeQueueUnBinding(const string &Ename, const string &Qname)
{
return _BMmp->UnBind(Ename, Qname);
}
// 判断绑定信息是否存在
bool ExistsBinding(const string &Ename, const string &Qname)
{
return _BMmp->Exists(Ename, Qname);
}
// 4、获取指定交换机的所有绑定信息
MsgQueuesBindingMap GetExchangeBindings(const string &Ename)
{
Exchange::ptr exp = _EXmp->SelectExchange(Ename);
if (exp == nullptr)
{
lg.LogMessage(error, "交换机绑定队列失败，交换机不存在！\n");
return MsgQueuesBindingMap();
}
return _BMmp->GetExchangeBindings(Ename);
}
// 5、消息的发布（新增）、获取队首消息（消费一条消息）、确认消息（确认后删除）
bool BasicPublishMessage(const string &Qname, MQ_message::BasicProperties *bp, const string &body)
{
MsgQueue::Queue_ptr mqp = _MQmp->SelectQueue(Qname);
if (mqp == nullptr)
{
lg.LogMessage(error, "发布新消息失败，消息队列不存在！\n");
return false;
}
return _MMmp->InsertMessage(Qname, bp, body, mqp->_durable);
}
ns_message::MessagePtr BasicConsumeMessage(const string &Qname)
{
return _MMmp->FrontMessage(Qname);
}
bool BasicAckQueueMessage(const string &Qname, const string &msg_id)
{
return _MMmp->AckQueueMessage(Qname, msg_id);
}
// 清理数据
void Clear()
{
_EXmp->ClearAllExchanges();
_MQmp->ClearAllQueues();
_BMmp->Clear();
_MMmp->Clear();
}
private:
ExchangeManager::ptr _EXmp;
MsgQueueManager::ptr _MQmp;
BindingManager::ptr _BMmp;
MessageManager::ptr _MMmp;
string _host_name; // 虚拟机名称，在本项目中由于简化操作，作用极小
};
}
1.3 路由匹配模块
用户发布信息时，实际上是先把消息发给交换机，由交换机决定把这些消息放到哪些队列中，不同类型的交换机，有着不同类型的匹配规则（广播交换、直接交换、主题交换），这个决定的过程就是路由匹配的过程。路由匹配模块就是实现此过程的。
基于功能需求的分析，路由交换模块只需要对传入的数据进行处理，因此这个模块实现的实际上是一个功能接口类，且没有成员变量
1.3.1 交换机类型介绍
RabbitMQ主要支持四种交换机类型Direct、Fanout、Topic、Header；其中Header这个类型比较复杂且少见，因此在RabbitMQ中常用的是前三种交换机类型，本项目也主要实现这三种：
1、Direct（直接交换）：绑定队列到交换机上时，指定一个字符串binding_key；发送消息指定一个字符串为routing_key。生产者发布消息时，routing_key和binding_key完全一致，则匹配成功（即发送消息时指明发送队列，一对一发送）
2、Fanout（广播交换）：生产者发布的消息会被直接复制到该交换机绑定的所有队列上
3、Header（主题交换）：当routing_key和binding_key满足一定的匹配条件（着重设置）时，把消息投递到指定队列
binding_key（发布队列的匹配规则）
binding_key是由数字字母下划线构成，并且使用 . 切割成若干部分。格式规定：指定由数字、字母、 _ 、# 、* 、构成
- 和 # 属于通配符，* 可以匹配任意一个单词（注意是单词，不是字母），# 则可以匹配任意0个/ 多个单词（同样注意是单词不是字母）。值得注意的是， * 和 # 两种通配符只能作为由 . 切分出来的独立部分，不可以从其他数字字母混用（eg：a .. b是合法的，a .a. b是不合法的）
routing_key（消息发布规则）
由数据、字母和下划线构成，并且使用 . 切割成若干部分（eg：news.sport.pop，表示这个消息是一个流行音乐的新闻）
本项目提供的路由匹配操作（具体实现在代码中）
1、判断routing_key和binding_key是否符合规定
2、提供判断routing_key和binding_key是否能够匹配成功的接口
（PS：本次的主题交换匹配规则：binding_key是xxx.xxx.#，当routing_key前两段符合，即为匹配成功）（eg：binding_key：news.music.#，routing_key：news.music.pop）
// 路由匹配：用户发布信息时，先把消息发给交换机，由交换机决定把这些消息放到哪些队列中，这个决定的过程就是路由匹配的过程
// 基于功能需求的分析，路由交换模块只需要对传入的数据进行处理，因此这个模块实现的实际上是一个功能接口类，并灭有成员变量
#pragma once
#include "../MQcommon/Helper.hpp"
#include "../MQcommon/Log.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include <iostream>
using namespace std;
using namespace ns_helper;
namespace ns_route
{
class Router
{
public:
// 判断routing_key是否合法，只需要判断是否包含有非法字符即可，合法字符( a~z, A~Z, 0~9, ., _)
static bool IsLegalRoutingKey(const string &routing_key)
{
for (auto &ch : routing_key)
{
if ((ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') || (ch >= '0' && ch <= '9') || (ch == '_' || ch == '.'))
{
continue;
}
return false;
}
return true;
}
// 判断binding_key是否合法
static bool IsLegalBindingKey(const string &binding_key)
{
// 1. 判断是否包含有非法字符， 合法字符：a~z, A~Z, 0~9, ., _, *, #
for (auto &ch : binding_key)
{
if ((ch >= 'a' && ch <= 'z') || (ch >= 'A' && ch <= 'Z') || (ch >= '0' && ch <= '9') || (ch == '_' || ch == '.') || (ch == '*' || ch == '#'))
{
continue;
}
return false;
}
// 2. *和#必须独立存在，通过以.切割字符串来判断: news.music#.*.#
vector<string> sub_words;
string sub_sep = ".";
StrHelper::SpiltStr(binding_key, sub_words, sub_sep);
for (string &word : sub_words)
{
// *和#必须独立存在于一个字符串中
if (word.size() > 1 && ((word.find("*") != std::string::npos) || (word.find("#") != std::string::npos)))
{
return false;
}
}
// 3. *和#不能连续出现
for (int i = 1; i < sub_words.size(); i++)
{
if (sub_words[i] == "#" && sub_words[i - 1] == "*")
{
return false;
}
if (sub_words[i] == "#" && sub_words[i - 1] == "#")
{
return false;
}
if (sub_words[i] == "*" && sub_words[i - 1] == "#")
{
return false;
}
}
return true;
}
// 进行路由匹配，传入参数：交换机交换类型，routing_key，binding_key
static bool Route(MQ_message::ExchangeType type, const string &routing_key, const string &binding_key)
{
// 直接交换，只要binding_key和routing_key匹配一致，即可交换
if (type == MQ_message::ExchangeType::direct)
{
return (routing_key == binding_key);
}
// 广播交换，无论如何，都是直接发布
else if (type == MQ_message::ExchangeType::fanout)
{
return true;
}
// 主题交换，按照匹配规则（动态规划）进行匹配，从数组的末端位置查询结果
else if (type == MQ_message::ExchangeType::topic)
{
// 1. 将binding_key与routing_key进行字符串分割，得到各个的单词数组
vector<string> bkeys, rkeys;
string sub_sep = ".";
size_t n_bkey = StrHelper::SpiltStr(binding_key, bkeys, sub_sep); // binding_key切割的单词数量
size_t n_rkey = StrHelper::SpiltStr(routing_key, rkeys, sub_sep); // routing_key切割的单词数量
/* printf("binding_key：");
for (int i = 0; i < bkeys.size(); i++)
{
cout << bkeys[i] << " ";
}
printf("\nrouting__key：");
for (int i = 0; i < rkeys.size(); i++)
{
cout << rkeys[i] << " ";
} */
cout << endl;
// 2. 按照元素个数，定义标记二维数组（额外申请1格，便于真正的第一行第一列继承左上方的结果），并初始化[0][0]位置为true，其他位置为false，
vector<vector<bool>> dp(n_bkey + 1, vector<bool>(n_rkey + 1, false));
dp[0][0] = true;
// 3. 如果binding_key以#起始，则将#对应行的第0列置为1，便于匹配
for (int i = 1; i <= bkeys.size(); i++)
{
if (bkeys[i - 1] == "#")
{
dp[i][0] = true;
continue;
}
break;
}
// 4. 使用routing_key中的每个单词与binding_key中的每个单词进行匹配并标记数组
for (int i = 1; i <= n_bkey; i++)
{
for (int j = 1; j <= n_rkey; j++)
{
// 如果当前bkey是个*，或者两个单词相同，表示单词匹配成功，则从左上方继承结果，只有前面都成功，这里才成功
if (bkeys[i - 1] == rkeys[j - 1] || bkeys[i - 1] == "*")
{
dp[i][j] = dp[i - 1][j - 1];
}
else if (bkeys[i - 1] == "#")
{
// 如果当前bkey是个#（任意1个/多个单词），则需要从左上，左边，上边继承结果（即从上一个与#匹配的地方继承结果，只要有1个成功就成功）
dp[i][j] = dp[i - 1][j - 1] | dp[i][j - 1] | dp[i - 1][j];
}
}
}
return dp[n_bkey][n_rkey];
}
else
{
lg.LogMessage(error, "匹配规则错误！不存在此规则！\n");
return false;
}
}
};
}
1.4 信道管理模块
信道是网络通信的一个概念，全称通信通道。
网络通信必然都是通过网络通信连接（Connection，后文也会实现）来完成的，网络通信连接是连接各种服务的基础，使得不同服务之间能够相互通信和交互。
但是为了更充分的利用资源和防止频繁开关网络连接，我们对网络连接进行了更一步的细化，细化出来了通信通道的概念；对于用户而言，通信通道才是网络通信的载体，每一个信道在用户眼里是相互独立的。（一个真正的通信连接可以创建出多个信道，但本质底层都是使用同一个网络连接进行网络通信）
因此，因为信道是用户眼中的一个通信通道，所以所有的网络通信服务都由信道提供。
一个连接可能会对应有多个通信通道，一旦某个客户端要关闭通信，关闭的不是连接，而是自己的信道；关闭信道的同时，要把消息队列的订阅也取消，这样生产者就不会再给消费者推送消息。
1.4.1 信道管理的数据
信道ID：信道的唯一标识
信道关联的消费者：用于消费者信道在关闭的时候取消订阅，删除订阅者信息
信道关联的连接：用于向客户端发送数据（响应、推送的消息）
protubuf协议处理句柄：用于网络通信前的协议处理
消费者管理句柄：信道关闭/取消订阅的时候，通过句柄删除订阅者信息
虚拟机句柄：集合了交换机、队列、绑定、消息数据管理
异步工作线程池句柄：信道进行消息发布，把信息推送到指定队列后，需要通知订阅该队列的订阅者进行消费，通知订阅该队列的消费者来消费消息的任务就被交给这个线程池来执行
1.4.2 信道管理的操作
1、声明 / 删除交换机
2、声明 / 删除队列
3、绑定 / 解绑交换机队列
4、发布消息 / 订阅队列消息 / 取消队列订阅 / 队列消息确认
// 信道管理模块，整合各项服务（本项目中一个信道只能有一个消费者）
#pragma once
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../MQcommon/mq_msg.pb.h"
#include "../MQcommon/mq_proto.pb.h"
#include "../MQcommon/Helper.hpp"
#include "../MQcommon/Log.hpp"
#include "../MQcommon/ThreadPool.hpp"
#include "Consumer.hpp"
#include "Route.hpp"
#include "VirtualHost.hpp"
#include <iostream>
using namespace std;
using namespace MQ_message;
using namespace ns_helper;
using namespace ns_consumer;
using namespace ns_route;
using namespace ns_VirtualHost;
using namespace ns_ThreadPool;
namespace ns_Channel
{
using ProtobufCodecPtr = shared_ptr<ProtobufCodec>;
// protobuf中，不同的请求类型，定义了不同的智能指针类型
using OpenChannelRequestPtr = shared_ptr<OpenChannelRequest>; // 打开信道
using CloseChannelRequestPtr = shared_ptr<CloseChannelRequest>; // 关闭信道
using DeclareExchangeRequestPtr = shared_ptr<DeclareExchangeRequest>; // 声明交换机
using DeleteExchangeRequestPtr = shared_ptr<DeleteExchangeRequest>; // 删除交换机
using DeclareQueueRequestPtr = shared_ptr<DeclareQueueRequest>; // 声明队列
using DeleteQueueRequestPtr = shared_ptr<DeleteQueueRequest>; // 删除队列
using QueueBindRequestPtr = shared_ptr<QueueBindRequest>; // 队列绑定
using QueueUnBindRequestPtr = shared_ptr<QueueUnBindRequest>; // 队列解绑
using BasicPublishRequestPtr = shared_ptr<BasicPublishRequest>; // 信息发布
using BasicAckRequestPtr = shared_ptr<BasicAckRequest>; // 信息确认
using BasicConsumeRequestPtr = shared_ptr<BasicConsumeRequest>; // 订阅信息
using BasicCancelRequestPtr = shared_ptr<BasicCancelRequest>; // 取消订阅
class Channel
{
public:
using ptr = shared_ptr<Channel>;
Channel(const string &Cid, const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec, const ConsumerManager::ptr &cmp, const VirtualHost::ptr &host, const ns_ThreadPool::ThreadPool::ptr &pool) : _Cid(Cid), _conn(conn), _codec(codec), _cmp(cmp), _host(host), _pool(pool)
{
}
~Channel()
{
// 信道关闭时，移除相关消费者
if (_consumer.get() != nullptr)
{
// 消费者存在
_cmp->RemoveConsumer(_consumer->_tag, _consumer->_Qname);
}
lg.LogMessage(info, "%s信道删除成功！\n", _Cid.c_str());
}
// 交换机的声明与删除（删除交换机的同时，交换机关联的绑定信息也要删除）
void DeclareExchange(const DeclareExchangeRequestPtr &rep)
{
bool res = _host->DeclareExchange(rep->exchange_name(), rep->exchange_type(), rep->durable(), rep->auto_delete(), rep->args());
BasicResponse(rep->rid(), rep->cid(), res);
}
void DeleteExchange(const DeleteExchangeRequestPtr &rep)
{
bool res = _host->DeleteExchange(rep->exchange_name());
BasicResponse(rep->rid(), rep->cid(), res);
}
// 队列的声明与删除
void DeclareQueue(const DeclareQueueRequestPtr &rep)
{
bool res = _host->DeclareQueue(rep->queue_name(), rep->durable(), rep->exclusive(), rep->auto_delete(), rep->args());
if (res == false)
{
BasicResponse(rep->rid(), rep->cid(), res);
return;
}
// 声明队列的同时，队列的消费者管理句柄也要初始化
_cmp->InitQueueConsumer(rep->queue_name());
BasicResponse(rep->rid(), rep->cid(), res);
}
void DeleteQueue(const DeleteQueueRequestPtr &rep)
{
bool res = _host->DeleteQueue(rep->queue_name());
// 删除队列的同时，删除队列关联的绑定信息、消息、消费者信息
_cmp->DestroyQueueConsumer(rep->queue_name()); // 删除消费者信息
_host->DeleteQueue(rep->queue_name()); // 整合了删除队列、删除队列消息和删除队列的绑定信息
BasicResponse(rep->rid(), rep->cid(), res);
}
// 队列的绑定与解绑
void BindingQueue(const QueueBindRequestPtr &rep)
{
bool res = _host->ExchangeQueueBinding(rep->exchange_name(), rep->queue_name(), rep->binding_key());
BasicResponse(rep->rid(), rep->cid(), res);
}
void UnBindingQueue(const QueueUnBindRequestPtr &rep)
{
bool res = _host->ExchangeQueueUnBinding(rep->exchange_name(), rep->queue_name());
BasicResponse(rep->rid(), rep->cid(), res);
}
// 消息的发布（哪一个信道调用这个函数，哪个信道连接的是生产者）
void BasicPublish(const BasicPublishRequestPtr &rep)
{
// 1、判断交换机是否存在
auto res = _host->SelectExchange(rep->exchange_name());
if (res == nullptr)
{
lg.LogMessage(info, "交换机不存在！消息发布失败\n");
return;
}
// 2、进行交换机路由匹配，判断消息可以发布到交换机绑定的哪个队列中
MsgQueuesBindingMap mqbm = _host->GetExchangeBindings(rep->exchange_name()); // 获取指定交换机的所有绑定信息
BasicProperties *properties = nullptr;
string routing_key;
// 判断消息属性是否存在
if (rep->has_properties())
{
properties = rep->mutable_properties();
routing_key = properties->routing_key();
}
for (auto &binding : mqbm)
{
if (Router::Route(res->_type, routing_key, binding.second->_binding_key))
{
// 3、发布消息（用于添加信息管理）
_host->BasicPublishMessage(binding.second->_msgqueue_name, properties, rep->body());
// 4、向线程池中添加一个信息消费任务，以便让该队列的消费者进行消费（用于向指定队列的订阅者推送消息——由线程池完成，主进程不做消息发布操作，因为消息发布需要网络传输，会阻塞当前操作，降低效率）
auto task = bind(&Channel::Consume, this, binding.first); // consume函数是个类内函数，不能直接传递，通过建立一个对象实例来传递
_pool->push(task);
}
}
BasicResponse(rep->rid(), rep->cid(), true);
}
// 消息的确认
void BasicAck(const BasicAckRequestPtr &rep)
{
bool res = _host->BasicAckQueueMessage(rep->queue_name(), rep->message_id());
BasicResponse(rep->rid(), rep->cid(), res);
}
// 队列消息的订阅与取消订阅（哪一个信道调用这两个函数，哪个信道连接的是消费者）
void BasicConsume(const BasicConsumeRequestPtr &rep)
{
// 1. 判断队列是否存在
bool res = _host->ExistsQueue(rep->queue_name());
if (res == false)
{
lg.LogMessage(error, "%s队列不存在，无法订阅！\n", rep->queue_name().c_str());
return;
}
// 2. 创建队列的消费者
auto cb = std::bind(&Channel::CallBack, this, placeholders::_1, placeholders::_2, placeholders::_3); // 绑定的this指针是谁，就给谁推送消息，这个this指针就是调用BasicConsume函数的对象
_consumer = _cmp->CreateConsumer(rep->consumer_tag(), rep->queue_name(), rep->auto_ack(), cb); // 创建了消费者之后，当前的channel角色就是个消费者
BasicResponse(rep->rid(), rep->cid(), res);
}
void BasicCancel(const BasicCancelRequestPtr &rep)
{
bool res = _cmp->RemoveConsumer(rep->consumer_tag(), rep->queue_name());
BasicResponse(rep->rid(), rep->cid(), res);
}
private:
// 针对参数组织出推送消息请求，将消息推送给channel对应的客户端
void CallBack(const string tag, const MQ_message::BasicProperties *bp, const string &body)
{
BasicConsumeResponse resp;
resp.set_cid(_Cid);
resp.set_consumer_tag(tag);
resp.set_body(body);
if (bp)
{
resp.mutable_properties()->set_id(bp->id());
resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
resp.mutable_properties()->set_routing_key(bp->routing_key());
}
_codec->send(_conn, resp); // 发送给客户端
}
// 通知订阅指定队列的消费者来消费消息
void Consume(const string &Qname)
{
// 1、从队列中取出一条消息
auto mp = _host->BasicConsumeMessage(Qname);
if (mp == nullptr)
{
lg.LogMessage(error, "%s队列消息获取失败！队列为空！\n", Qname.c_str());
return;
}
// 2、从队列订阅者中取出一个订阅者（不能用与信道关联的消费者，因为此处是指定队列的消费者消费消息）
auto cp = _cmp->ChooseConsumer(Qname);
if (cp == nullptr)
{
lg.LogMessage(error, "%s队列消息消费失败！没有消费者！\n", Qname.c_str());
return;
}
// 3、调用订阅者对应的消息处理回调函数，实现消息的推送，这个回调函数已经通过this指针绑定了消费者，调用它等于传递给这个消费者消费
cp->_callback(cp->_tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
// 4、判断如果订阅者是自动确认---不需要等待确认，直接删除消息，否则需要外部收到消息确认后再删除
if (cp->_auto_ack)
{
_host->BasicAckQueueMessage(Qname, mp->payload().properties().id());
}
}
// 通用返回响应结果
void BasicResponse(const string &rid, const string &cid, bool ok)
{
BasicCommonResponse rep;
rep.set_rid(rid);
rep.set_cid(cid);
rep.set_ok(ok);
_codec->send(_conn, rep); // 把数据序列化后发送出去
}
string _Cid; // 信道ID，信道的唯一标识
Consumer::ptr _consumer; // 信道关联的消费者：用于消费者信道在关闭的时候取消订阅，删除订阅者信息
muduo::net::TcpConnectionPtr _conn; // 信道关联的连接：用于向客户端发送数据（响应、推送的消息）
ProtobufCodecPtr _codec; // protubuf协议处理句柄：用于网络通信前的协议处理
ConsumerManager::ptr _cmp; // 消费者管理句柄：信道关闭/取消订阅的时候，通过句柄删除订阅者信息
VirtualHost::ptr _host; // 虚拟机句柄：集合了交换机、队列、绑定、消息数据管理
ns_ThreadPool::ThreadPool::ptr _pool; // 异步工作线程池句柄
};
// 信道管理句柄，完成信道的增删查改
class ChannelManager
{
public:
using ptr = shared_ptr<ChannelManager>;
ChannelManager()
{
}
// 打开信道
bool OpenChannel(const string &Cid, const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec, const ConsumerManager::ptr &cmp, const VirtualHost::ptr &host, const ns_ThreadPool::ThreadPool::ptr &pool)
{
unique_lock<mutex> lock(_mutex);
auto it = _channels.find(Cid);
if (it != _channels.end())
{
lg.LogMessage(warning, "%s信道已打开！不需要再次打开\n", Cid.c_str());
return false;
}
auto channel = make_shared<Channel>(Cid, conn, codec, cmp, host, pool);
_channels[Cid] = channel;
return true;
}
// 关闭信道
bool CloseChannel(const string &Cid)
{
unique_lock<mutex> lock(_mutex);
auto it = _channels.find(Cid);
if (it == _channels.end())
{
lg.LogMessage(warning, "%s信道已关闭！\n", Cid.c_str());
return false;
}
_channels.erase(Cid);
return true;
}
// 获取某个信道
Channel::ptr GetChannel(const string &Cid)
{
unique_lock<mutex> lock(_mutex);
auto it = _channels.find(Cid);
if (it == _channels.end())
{
lg.LogMessage(warning, "%s信道不存在！\n", Cid.c_str());
return nullptr;
}
return it->second;
}
private:
mutex _mutex;
unordered_map<string, Channel::ptr> _channels;
};
}
1.5 连接管理模块
在网络通信模块中，我们使用muduo库来实现底层的通信，muduo库本身就有Connection连接有关的概念和对象类 ，但是muduo库中的Connection连接中并没有信道的概念。
因此，我们需要在用户层面，对muduo库中的Connection进行二次封装，形成我们自己所需要的连接管理。
（当一个网络通信对应的连接关闭的时候，应该把该连接关联的信道全部关闭，并取消订阅客户端的订阅，因此也必须要有数据管理，至少要了解某一连接关联了哪些信道）
1.5.1 连接管理的数据
1、连接管理的所有信道
信道需要的句柄都要由连接来提供，因为信道是通过连接来创建的，因此信道需要什么资源，连接就要提供什么资源（连接中的这些资源将从服务端获得）
2、信道关联的muduo库中的Conenction连接
3、protubuf协议处理句柄：用于网络通信前的协议处理
4、 消费者管理句柄：信道关闭/取消订阅的时候，通过句柄删除订阅者信息
5、虚拟机句柄：集合了交换机、队列、绑定、消息数据管理
6、异步工作线程池句柄
1.5.2 连接提供的操作
1、打开 / 关闭信道
2、创建 / 关闭连接
3、获取某一信道 / 某一连接的信息
// 连接管理，用于与客户端进行网络通信
#include "Channel.hpp"
using namespace std;
using namespace ns_Channel;
namespace ns_Connection
{
class Connection
{
public:
using ptr = shared_ptr<Connection>;
Connection(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec, const ConsumerManager::ptr &cmp, const VirtualHost::ptr &host, const ThreadPool::ptr &pool)
: _conn(conn), _codec(codec), _cmp(cmp), _host(host), _pool(pool), _channels(std::make_shared<ChannelManager>())
{
}
/* 不需要析构函数，使用默认即可，因为智能指针会随着对象生命周期自动释放
~Connection()
{
// 析构时，要把其关联的所有信道全部关闭
}
*/
// 打开信道
void OpenChannel(const OpenChannelRequestPtr &rep)
{
// 1、判断某ID信道是否已经打开（信道ID是信道的唯一标识，不可重复），并利用信道管理句柄打开信道
bool res = _channels->OpenChannel(rep->cid(), _conn, _codec, _cmp, _host, _pool);
if (res == false)
{
BasicResponse(rep->rid(), rep->cid(), false);
return;
}
// 2、给客户端的每一个请求进行回复
lg.LogMessage(info, "%s 信道创建成功！\n", rep->cid().c_str());
BasicResponse(rep->rid(), rep->cid(), true);
}
// 关闭信道
void CloseChannel(const CloseChannelRequestPtr &rep)
{
bool res = _channels->CloseChannel(rep->cid());
if (res == false)
{
BasicResponse(rep->rid(), rep->cid(), false);
return;
}
BasicResponse(rep->rid(), rep->cid(), true);
}
// 获取信道信息
Channel::ptr GetChannel(const string &Cid)
{
return _channels->GetChannel(Cid);
}
private:
// 通用返回响应结果
void BasicResponse(const string &rid, const string &cid, bool ok)
{
BasicCommonResponse rep;
rep.set_rid(rid);
rep.set_cid(cid);
rep.set_ok(ok);
_codec->send(_conn, rep); // 把数据序列化后发送出去
}
ChannelManager::ptr _channels; // 连接管理的所有信道
// 信道需要的句柄都要由连接来提供，因为信道是通过连接来创建的，信道需要什么资源，连接就要提供什么资源（连接中的这些资源将从服务端获得）
muduo::net::TcpConnectionPtr _conn; // 信道关联的连接：用于向客户端发送数据（响应、推送的消息）
ProtobufCodecPtr _codec; // protubuf协议处理句柄：用于网络通信前的协议处理
ConsumerManager::ptr _cmp; // 消费者管理句柄：信道关闭/取消订阅的时候，通过句柄删除订阅者信息
VirtualHost::ptr _host; // 虚拟机句柄：集合了交换机、队列、绑定、消息数据管理
ThreadPool::ptr _pool; // 异步工作线程池句柄
};
// 连接的管理
class ConnectionManager
{
public:
using ptr = shared_ptr<ConnectionManager>;
ConnectionManager()
{
}
// 新建连接
void NewConnection(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec, const ConsumerManager::ptr &cmp, const VirtualHost::ptr &host, const ThreadPool::ptr &pool)
{
unique_lock<mutex> lock(_mutex);
auto it = _connections.find(conn);
if (it != _connections.end())
{
lg.LogMessage(warning, "连接已建立！不需要再次建立\n");
return;
}
Connection::ptr _conn = make_shared<Connection>(conn, codec, cmp, host, pool);
_connections[conn] = _conn;
}
// 删除连接
void DeleteConnection(const muduo::net::TcpConnectionPtr &conn)
{
unique_lock<mutex> lock(_mutex);
auto it = _connections.find(conn);
if (it == _connections.end())
{
lg.LogMessage(warning, "连接已删除！不需要重复删除\n");
return;
}
_connections.erase(conn);
}
// 获取连接
Connection::ptr GetConnection(const muduo::net::TcpConnectionPtr &conn)
{
unique_lock<mutex> lock(_mutex);
auto it = _connections.find(conn);
if (it == _connections.end())
{
lg.LogMessage(warning, "连接不存在！\n");
return nullptr;
}
return _connections[conn];
}
private:
mutex _mutex;
unordered_map<muduo::net::TcpConnectionPtr, Connection::ptr> _connections; // Muduo库的连接与自己建立的连接的映射集合
};
}
1.6 消费者管理模块
客户端有两种：发布消息的客户端、订阅消息的客户端，订阅了某个队列的消息的客户端就是消费者。核心接口里有个订阅消息接口，消费者订阅某个消息队列，当这个队列有消息后，队列所连接的服务端会将队列消息轮询推送给订阅这个队列的消费者
因为每个消费者都会订阅某个消息队列，所以消费者管理模块是以队列为单元进行的。
（PS：在本项目中，服务端负责将消息推送给订阅者，订阅者以推模式接收信息，即订阅者客户端并不主动拉取消息，而是订阅某一消息队列，服务端把消息推送给订阅者。因此操作流程通常是——服务端从关联的消息队列里取出消息，再从订阅该队列的消费者中取出一个（采用RR轮转算法选择），将消息推送给它）
1.6.1 管理的消费者数据
1、消费者标识
2、订阅的消息队列名称
3、是否自动应答标志（若为真，一个消息被消费者取走消费后，消息会被自动确认，待确认消息表会直接移除该待确认信息；若为假，则需要等待客户端确认）
4、消息的回调函数：队列有了消息后，通过该函数进行处理，服务端调用这个回调函数，**内部逻辑是服务端找到消费者对应的连接，然后将数据发送给消费者对应的客户端（在信道模块实现）**
1.6.2 管理的消费者操作
** **1、客户端订阅者指定队列消息时，向指定队列新增消费者，新增完成后返回消费者对象
2、RR轮转算法获取某一队列消费者
3、获取订阅某一队列的消费者数量
4、判断指定的消费者是否存在
5、客户端取消订阅时，从指定队列中移除消费者（移除哪个队列的哪个消费者）
// 消费者（订阅者）信息管理模块
// 不需要持久化存储，因为信道与连接相关，连接断开，信道自然断开
#pragma once
#include "../MQcommon/Helper.hpp"
#include "../MQcommon/Log.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <vector>
#include <mutex>
#include <memory>
#include <functional>
using namespace std;
using namespace ns_helper;
namespace ns_consumer
{
// 消费者（订阅者）结构
using CallBack = function<void(const string, const MQ_message::BasicProperties *bp, const string)>;
class Consumer
{
public:
using ptr = shared_ptr<Consumer>;
Consumer()
{
lg.LogMessage(info, "新增消费者：%p\n", this);
}
Consumer(const string &tag, const string &Qname, bool auto_ack, const CallBack &callback) : _tag(tag), _Qname(Qname), _auto_ack(auto_ack), _callback(callback)
{
lg.LogMessage(info, "新增消费者：%p\n", this);
}
~Consumer()
{
lg.LogMessage(info, "消费者 %p 已退出！\n", this);
}
string _tag; // 消费者标识
string _Qname; // 订阅的消息队列名称
bool _auto_ack; // 是否自动应答标志（若为真，一个消息被消费者消费后，会直接移除该待确认信息；若为假，需要等待客户端确认）
CallBack _callback; // 消息的回调函数（服务端调用这个回调函数，内部逻辑是服务端找到消费者对应的连接，然后将数据发送给消费者对应的客户端）
};
// 以队列为单元进行管理的消费者管理结构——队列消费者管理结构
// 为什么以队列为单元？因为大部分使用情况是消息队列收到消息后，去找到订阅它的消费者，而非信道关闭时去删除消费者
class QueueConsumer
{
public:
using ptr = shared_ptr<QueueConsumer>;
QueueConsumer(const string &Qname)
{
}
// 创建消费者：信道提供的服务是在订阅队列消息的时候，创建消费者
Consumer::ptr Create(const string &tag, const string &Qname, bool auto_ack, const CallBack &callback)
{
// 1、加锁
unique_lock<mutex> lock(_mutex);
// 2、判断消费者是否重复
for (auto consumer : _consumers)
{
if (consumer->_tag == tag)
{
lg.LogMessage(warning, "消费者重复！\n");
return nullptr;
}
}
// 3、没有重复则构造对象，新增消费者
auto consumer = make_shared<Consumer>(tag, Qname, auto_ack, callback);
// 4、添加进管理结构后返回对象
_consumers.push_back(consumer);
return consumer;
}
// 移除消费者：取消订阅/信道关闭/连接关闭的时候，移除消费者
bool Remove(const string &tag)
{
// 1、加锁
unique_lock<mutex> lock(_mutex);
// 2、查找并删除，返回结果
for (auto it = _consumers.begin(); it != _consumers.end(); it++)
{
if ((*it)->_tag == tag)
{
_consumers.erase(it);
return true;
}
}
return false;
}
// 选择消费者：从队列中所有的消费者中按RR轮转顺序，取出一个消费者用于消息的推送（消费）
Consumer::ptr Choose()
{
// 1、加锁
unique_lock<mutex> lock(_mutex);
if (_consumers.size() == 0)
{
return Consumer::ptr();
}
// 2、获取当前轮转的下标
int num = RR_num % _consumers.size();
RR_num++;
// 3、获取对象并返回
return _consumers[num];
}
// 判断队列消费者是否为空
bool Empty()
{
unique_lock<mutex> lock(_mutex);
return _consumers.size() == 0;
}
// 判断指定的消费者是否存在
bool Exists(const string &tag)
{
// 1、加锁
unique_lock<mutex> lock(_mutex);
// 2、判断消费者是否重复
for (auto consumer : _consumers)
{
if (consumer->_tag == tag)
{
return true;
}
}
return false;
}
// 清理队列的所有消费者
void Clear()
{
unique_lock<mutex> lock(_mutex);
_consumers.clear();
RR_num = 0;
}
private:
string _Qname; // 队列名称
vector<Consumer::ptr> _consumers; // 消费者管理结构
size_t RR_num; // 一个队列可能会有多个消费者，但是一条消息只需要被一个消费者消费，因此采用RR轮转顺序
mutex _mutex; // 互斥锁，保证线程安全
};
// 向外提供的消费者统一管理句柄
class ConsumerManager
{
public:
using ptr = shared_ptr<ConsumerManager>;
ConsumerManager()
{
}
// 初始化/删除队列的消费者信息管理结构（创建/删除队列的时候初始化）
bool InitQueueConsumer(const string &Qname)
{
// 1、加锁
unique_lock<mutex> lock(_mutex);
// 2、重复判断
auto it = _Qconsumers.find(Qname);
if (it != _Qconsumers.end())
{
// lg.LogMessage(warning, "队列重复！\n");
return false;
}
// 3、新增
auto consumer = make_shared<QueueConsumer>(Qname);
_Qconsumers[Qname] = consumer;
return true;
}
void DestroyQueueConsumer(const string &Qname)
{
unique_lock<mutex> lock(_mutex);
_Qconsumers.erase(Qname);
}
// 客户端订阅者指定队列消息时，向指定队列新增消费者，新增完成后返回消费者对象
Consumer::ptr CreateConsumer(const string &tag, const string &Qname, bool auto_ack, const CallBack &callback)
{
{
unique_lock<mutex> lock(_mutex); // 这个锁限制作用域，仅用来寻找对应的队列消费者管理句柄
auto it = _Qconsumers.find(Qname);
if (it == _Qconsumers.end())
{
lg.LogMessage(warning, "未找到%s队列的消费者管理句柄\n", Qname.c_str());
return nullptr;
}
}
return _Qconsumers[Qname]->Create(tag, Qname, auto_ack, callback);
}
// 客户端取消订阅时，从指定队列中移除消费者（移除哪个队列的哪个消费者）
bool RemoveConsumer(const string &tag, const string &Qname)
{
{
unique_lock<mutex> lock(_mutex); // 这个锁限制作用域，仅用来寻找对应的队列消费者管理句柄
auto it = _Qconsumers.find(Qname);
if (it == _Qconsumers.end())
{
lg.LogMessage(warning, "未找到%s队列的消费者管理句柄\n", Qname.c_str());
return false;
}
}
return _Qconsumers[Qname]->Remove(tag);
}
// 从指定队列获取一个消费者（轮询获取-消费者轮换消费，起到负载均衡的作用）
Consumer::ptr ChooseConsumer(const string &Qname)
{
{
unique_lock<mutex> lock(_mutex); // 这个锁限制作用域，仅用来寻找对应的队列消费者管理句柄
auto it = _Qconsumers.find(Qname);
if (it == _Qconsumers.end())
{
lg.LogMessage(warning, "未找到%s队列的消费者管理句柄\n", Qname.c_str());
return nullptr;
}
}
return _Qconsumers[Qname]->Choose();
}
// 判断指定队列中消费者是否为空
bool Empty(const string &Qname)
{
{
unique_lock<mutex> lock(_mutex); // 这个锁限制作用域，仅用来寻找对应的队列消费者管理句柄
auto it = _Qconsumers.find(Qname);
if (it == _Qconsumers.end())
{
lg.LogMessage(warning, "未找到%s队列的消费者管理句柄\n", Qname.c_str());
return false;
}
}
return _Qconsumers[Qname]->Empty();
}
// 判断指定队列的指定消费者是否存在
bool Exists(const string &tag, const string &Qname)
{
{
unique_lock<mutex> lock(_mutex); // 这个锁限制作用域，仅用来寻找对应的队列消费者管理句柄
auto it = _Qconsumers.find(Qname);
if (it == _Qconsumers.end())
{
lg.LogMessage(warning, "未找到%s队列的消费者管理句柄\n", Qname.c_str());
return false;
}
}
return _Qconsumers[Qname]->Exists(tag);
}
// 清理所有消费者
void Clear()
{
unique_lock<mutex> lock(_mutex);
_Qconsumers.clear();
}
private:
mutex _mutex; // 保证_Qconsumers的线程安全
unordered_map<string, QueueConsumer::ptr> _Qconsumers; // 队列名和与其相关的消费者管理结构映射关系的集合
};
}
1.7 Server服务器模块
该模块是一个功能整合模块，并不亲自进行什么实质性的功能操作，而是接收客户端的各种操作请求，根据请求的不同，分发给上述各个模块执行；该模块最重要的是进行各功能的整合，是一个资源的载体。（本质上是对muduo库中的TcpServer类的二次封装与扩展）
整合的各大数据
1、epoll的事件监控：会进行描述符的事件监控，触发事件后进行io操作，业务处理
2、请求分发器对象：要向其中注册请求处理函数，根据请求的不同，调用不同的处理函数
3、muduo库基本服务器对象TcpServer，主要用于设置回调函数，用于告诉服务器收到什么请求该如何处理
4、protobuf协议处理器：对收到的请求数据进行protobuf协议处理
5、服务器中的消费者信息管理句柄
6、异步工作线程池：主要用于对队列消息的推送工作
7、连接管理句柄：管理当前服务器上的所有已经建立的通信连接
8、服务器持有的虚拟主机：队列、交换机、绑定、消息等数据都是通过虚拟主机来管理
整合的各个处理请求
1、打开信道、关闭信道请求
2、声明交换机、删除交换机请求*（思路：先找连接、再找信道、然后再操作交换机）*
3、声明队列请求
4、绑定交换机 - 队列请求
*5、删除队列请求*（删除队列前，移除指定队列的相关绑定信息数据）*** *6、发布消息请求
7、确认消息请求
8、订阅消息队列请求
9、取消订阅消息队列
这些请求全部都被提前定义在mq_proto.proto文件中，便于protobuf协议化处理和网络传输：
// 各个模块的protobuf协议定制
syntax = "proto3";
package MQ_message;
import "mq_msg.proto";
// 信道的打开请求
message OpenChannelRequest
{
string rid = 1; // 请求ID
string cid = 2; // 信道ID，信道的唯一标识
};
// 信道的关闭请求
message CloseChannelRequest
{
string rid = 1;
string cid = 2;
};
//交换机的声明与删除
message DeclareExchangeRequest
{
string rid = 1;
string cid = 2;
string exchange_name = 3; // 交换机名称
ExchangeType exchange_type = 4; // 交换机类型
bool durable = 5; // 交换机是否持久化标志
bool auto_delete = 6; // 是否自动删除标志
map<string, string> args = 7; // 其它参数
};
message DeleteExchangeRequest
{
string rid = 1;
string cid = 2;
string exchange_name = 3;
};
//队列的声明与删除
message DeclareQueueRequest
{
string rid = 1;
string cid = 2;
string queue_name = 3; // 队列名称
bool exclusive = 4; // 是否独占标志（同一时间内只能有一个线程可以访问该队列）
bool durable = 5; // 队列是否持久化标志
bool auto_delete = 6; // 是否自动删除标志
map<string, string> args = 7; // 其他参数
};
message DeleteQueueRequest
{
string rid = 1;
string cid = 2;
string queue_name = 3;
};
//队列的绑定与解除绑定
message QueueBindRequest
{
string rid = 1;
string cid = 2;
string exchange_name = 3; // 交换机名称
string queue_name = 4; // 队列名称
string binding_key = 5; // 分发匹配规则，决定哪些数据可以被从交换机放入队列
};
message QueueUnBindRequest
{
string rid = 1;
string cid = 2;
string exchange_name = 3;
string queue_name = 4;
};
//消息的发布
message BasicPublishRequest
{
string rid = 1;
string cid = 2;
string exchange_name = 3; // 交换机名称
string body = 4; // 消息正文内容
BasicProperties properties = 5; // 消息属性的主体部分
};
//消息的确认
message BasicAckRequest
{
string rid = 1;
string cid = 2;
string queue_name = 3; // 发布给哪个消息队列
string message_id = 4; // 消息ID（哪个队列的哪一条消息）
};
//队列的订阅
message BasicConsumeRequest
{
string rid = 1;
string cid = 2;
string consumer_tag =3; // 消费者ID标识
string queue_name = 4; // 订阅的消息队列名称
bool auto_ack = 5; // 是否自动应答标志（若为真，一个消息被消费者消费后，会直接移除该待确认信息；若为假，需要等待客户端确认）
};
//订阅的取消
message BasicCancelRequest
{
string rid = 1;
string cid = 2;
string consumer_tag = 3;
string queue_name = 4;
};
//消息的推送
message BasicConsumeResponse
{
string cid = 1;
string consumer_tag = 2;
string body = 3; // 消息正文内容
BasicProperties properties = 4; // 消息属性的主体部分
};
//通用响应
message BasicCommonResponse
{
string rid = 1;
string cid = 2;
bool ok = 3; // 响应是否OK
}
下面是具体实现代码
// 服务器模块，整合各大模块
#pragma once
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"
#include "../MQcommon/mq_msg.pb.h"
#include "../MQcommon/mq_proto.pb.h"
#include "../MQcommon/ThreadPool.hpp"
#include "../MQcommon/Log.hpp"
#include <iostream>
#include "Connection.hpp"
#include "Consumer.hpp"
#include "VirtualHost.hpp"
using namespace std;
using namespace ns_message;
using namespace ns_ThreadPool;
using namespace ns_Connection;
using namespace ns_consumer;
using namespace ns_Channel;
namespace ns_Server
{
#define DBFILE "/meta.db"
#define HOSTNAME "MyVirtualHost"
using MessagePtr = shared_ptr<google::protobuf::Message>;
// protobuf中，不同的请求类型，定义了不同的智能指针类型
using OpenChannelRequestPtr = shared_ptr<OpenChannelRequest>; // 打开信道
using CloseChannelRequestPtr = shared_ptr<CloseChannelRequest>; // 关闭信道
using DeclareExchangeRequestPtr = shared_ptr<DeclareExchangeRequest>; // 声明交换机
using DeleteExchangeRequestPtr = shared_ptr<DeleteExchangeRequest>; // 删除交换机
using DeclareQueueRequestPtr = shared_ptr<DeclareQueueRequest>; // 声明队列
using DeleteQueueRequestPtr = shared_ptr<DeleteQueueRequest>; // 删除队列
using QueueBindRequestPtr = shared_ptr<QueueBindRequest>; // 队列绑定
using QueueUnBindRequestPtr = shared_ptr<QueueUnBindRequest>; // 队列解绑
using BasicPublishRequestPtr = shared_ptr<BasicPublishRequest>; // 信息发布
using BasicAckRequestPtr = shared_ptr<BasicAckRequest>; // 信息确认
using BasicConsumeRequestPtr = shared_ptr<BasicConsumeRequest>; // 订阅信息
using BasicCancelRequestPtr = shared_ptr<BasicCancelRequest>; // 取消订阅
class Server
{
public:
// 传入的basedir带有路径
Server(int port, const string &basedir) : _server(&_loop, muduo::net::InetAddress("0.0.0.0" /*可以监控本机所有IP地址*/, port), "server", muduo::net::TcpServer::kReusePort),
_dispatcher(bind(&Server::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
/*请求分发处理函数，其中有个表，注册着不同业务的请求处理函数*/
_codec(make_shared<ProtobufCodec>(bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, placeholders::_1, placeholders::_2, placeholders::_3))),
_consumer(make_shared<ConsumerManager>()),
_threadpool(make_shared<ns_ThreadPool::ThreadPool>()),
_connections(make_shared<ConnectionManager>()),
_host(make_shared<VirtualHost>(HOSTNAME, basedir, basedir + DBFILE))
{
// 初始化历史消息中的所有队列，同时别忘了初始化队列的消费者管理结构
unordered_map<string, MsgQueue::Queue_ptr> queues = _host->GetAllQueues();
for (auto queue : queues)
{
_consumer->InitQueueConsumer(queue.first);
}
// 注册不同业务请求处理函数，根据收到的请求不同，选择不同的回调函数处理
_dispatcher.registerMessageCallback<OpenChannelRequest>(bind(&Server::OnOpenChannel, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<CloseChannelRequest>(bind(&Server::OnCloseChannel, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<DeclareExchangeRequest>(bind(&Server::OnDeclareExchange, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<DeleteExchangeRequest>(bind(&Server::OnDeleteExchange, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<DeclareQueueRequest>(bind(&Server::OnDeclareQueue, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<DeleteQueueRequest>(bind(&Server::OnDeleteQueue, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<QueueBindRequest>(bind(&Server::OnQueueBind, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<QueueUnBindRequest>(bind(&Server::OnQueueUnBind, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<BasicPublishRequest>(bind(&Server::OnBasicPublish, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<BasicAckRequest>(bind(&Server::OnBasicAck, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<BasicConsumeRequest>(bind(&Server::OnBasicConsume, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<BasicCancelRequest>(bind(&Server::OnBasicCancel, this, placeholders::_1, placeholders::_2, placeholders::_3));
// 设置消息回调处理函数，对收到的数据进行协议解析，一旦服务器收到数据，就会调用对应函数进行处理
/*对接收到的消息进行protobuf协议处理的函数，内部处理完毕后，会得到一个protobuf定义的请求结构对象，然后调用分发函数onProtobufMessage进行不同请求的处理*/
_server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), placeholders::_1, placeholders::_2, placeholders::_3));
// 连接建立成功/断开后处理函数
_server.setConnectionCallback(std::bind(&Server::onConnection, this, placeholders::_1));
}
void start()
{
_server.start(); // 服务器启动监听
_loop.loop(); // 启动事件监控，业务处理
}
private:
// 找到连接，操作信道
// 打开信道请求
void OnOpenChannel(const muduo::net::TcpConnectionPtr &conn, const OpenChannelRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "打开信道失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
connection->OpenChannel(message);
}
// 关闭信道请求
void OnCloseChannel(const muduo::net::TcpConnectionPtr &conn, const CloseChannelRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "关闭信道失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
connection->CloseChannel(message);
}
// 先找连接，再找信道，然后操作交换机
// 声明交换机请求
void OnDeclareExchange(const muduo::net::TcpConnectionPtr &conn, const DeclareExchangeRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "声明交换机失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
Channel::ptr channel = connection->GetChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(warning, "声明交换机失败！没有找到对应信道！\n");
return;
}
channel->DeclareExchange(message);
}
// 删除交换机请求
void OnDeleteExchange(const muduo::net::TcpConnectionPtr &conn, const DeleteExchangeRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "删除交换机失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
Channel::ptr channel = connection->GetChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(warning, "删除交换机失败！没有找到对应信道！\n");
return;
}
channel->DeleteExchange(message);
}
// 找连接 -> 找信道 -> 操作队列（交换机名称在请求中都已包含）
// 声明队列请求
void OnDeclareQueue(const muduo::net::TcpConnectionPtr &conn, const DeclareQueueRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "声明队列失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
Channel::ptr channel = connection->GetChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(warning, "声明队列失败！没有找到对应信道！\n");
return;
}
channel->DeclareQueue(message);
}
// 删除队列请求
void OnDeleteQueue(const muduo::net::TcpConnectionPtr &conn, const DeleteQueueRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "删除队列失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
Channel::ptr channel = connection->GetChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(warning, "删除队列失败！没有找到对应信道！\n");
return;
}
channel->DeleteQueue(message);
}
// 队列绑定请求
void OnQueueBind(const muduo::net::TcpConnectionPtr &conn, const QueueBindRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "绑定队列失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
Channel::ptr channel = connection->GetChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(warning, "绑定队列失败！没有找到对应信道！\n");
return;
}
channel->BindingQueue(message);
}
// 队列解绑请求
void OnQueueUnBind(const muduo::net::TcpConnectionPtr &conn, const QueueUnBindRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "解绑队列失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
Channel::ptr channel = connection->GetChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(warning, "解绑队列失败！没有找到对应信道！\n");
return;
}
channel->UnBindingQueue(message);
}
// 信息发布请求
void OnBasicPublish(const muduo::net::TcpConnectionPtr &conn, const BasicPublishRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "信息发布失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
Channel::ptr channel = connection->GetChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(warning, "信息发布失败！没有找到对应信道！\n");
return;
}
channel->BasicPublish(message);
}
// 信息确认请求
void OnBasicAck(const muduo::net::TcpConnectionPtr &conn, const BasicAckRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "信息确认失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
Channel::ptr channel = connection->GetChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(warning, "信息确认失败！没有找到对应信道！\n");
return;
}
channel->BasicAck(message);
}
// 队列订阅信息请求
void OnBasicConsume(const muduo::net::TcpConnectionPtr &conn, const BasicConsumeRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "队列订阅信息失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
Channel::ptr channel = connection->GetChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(warning, "队列订阅信息失败！没有找到对应信道！\n");
return;
}
channel->BasicConsume(message);
}
// 队列消息取消订阅请求
void OnBasicCancel(const muduo::net::TcpConnectionPtr &conn, const BasicCancelRequestPtr &message, muduo::Timestamp)
{
Connection::ptr connection = _connections->GetConnection(conn);
if (connection.get() == nullptr)
{
lg.LogMessage(warning, "队列消息取消订阅失败！没有找到连接对应的Connection对象！TCP连接即将关闭。\n");
conn->shutdown();
return;
}
Channel::ptr channel = connection->GetChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(warning, "队列消息取消订阅失败！没有找到对应信道！\n");
return;
}
channel->BasicCancel(message);
}
// muduo库中没有的请求，如何处理
void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
{
LOG_INFO << "未知请求: " << message->GetTypeName() << "\n";
conn->shutdown();
}
// 处理连接成功/失败
void onConnection(const muduo::net::TcpConnectionPtr &conn)
{
if (conn->connected())
{
_connections->NewConnection(conn, _codec, _consumer, _host, _threadpool);
lg.LogMessage(info, "成功与服务端建立连接！");
}
else
{
_connections->DeleteConnection(conn);
LOG_INFO << "连接已断开！\n";
}
}
muduo::net::EventLoop _loop; // epoll的事件监控，会进行描述符的事件监控，触发事件后进行io操作，业务处理
ProtobufDispatcher _dispatcher; // 请求分发器对象，要向其中注册请求处理函数，根据请求的不同，调用不同的处理函数
muduo::net::TcpServer _server; // 基本服务器对象，主要用于设置回调函数，用于告诉服务器收到什么请求该如何处理
ProtobufCodecPtr _codec; // protobuf协议处理器，对收到的请求数据进行protobuf协议处理
ConsumerManager::ptr _consumer; // 服务器中的消费者信息管理句柄
ns_ThreadPool::ThreadPool::ptr _threadpool; // 异步工作线程池，主要用于队列消息的推送工作
ConnectionManager::ptr _connections; // 连接管理句柄，管理当前服务器上的所有已经建立的通信连接
VirtualHost::ptr _host; // 服务器持有的虚拟主机，队列、交换机、绑定、消息等数据都是通过虚拟主机来管理
};
}
2、客户端模块
客户端分为订阅客户端和发布客户端
订阅客户端：订阅一个消息队列，处理服务端推送而来的消息（订阅客户端每订阅一个消息队列的消息，就相当于创建了一个消费者）
发布客户端：向某个客户端发布消息
2.1 信道模块（信道请求模块）
信道模块是客户端模块的核心子模块，对于客户端而言，所有的服务都是通过信道完成的。一个直接面向用户的模块，内部包含多个向外提供的服务接口，用户需要什么服务，就调用对应的服务接口
客户端信道与服务端信道是一一对应的，服务端信道所提供的服务，客户端也都有，非常类似。相当于服务端为客户端提供服务，客户端为用户提供服务。
服务流程可以简单概括为：用户调用客户端所需服务的服务接口（存放在信道模块中）发布请求，客户端通过信道把请求发给服务端，服务端通过其信道获取请求，处理后返回响应。
2.1.1 信道管理的数据
1、信道ID
2、信道关联的网络通信连接对象
3、protobuf协议处理对象
4、信道关联的消费者*（不一定存在，发布客户端是没有的，因为发布客户端是生产者）*
5、请求ID对应的通用响应队列映射*（只要收到的不是推送消息的响应，都放入此队列中，便于快速判断响应）*
2.2.2 信道模块提供的接口
1、声明 / 删除交换机
2、声明 / 删除队列
3、绑定 / 解绑交换机队列
4、发布消息（客户端发布消息时，不需要指定队列名称，只需要指定交换机名称，发布给交换机即可，具体放到哪个队列由交换机决定，**只有发布者需要用到此接口**）
5、队列消息确认（确认收到哪一个队列的哪一条消息，**只有订阅者需要用到此接口**）
6、队列的订阅（需要额外传入收到来自服务端的推送消息的响应后的业务处理回调函数，**只有订阅者需要用到此接口**）
7、订阅的取消（订阅后，由关联的消费者提供参数，不需要用户给予，也是**只有订阅者需要用到此接口**）
8、处理消息接口：连接收到来自服务端的推送消息的响应后，需要通过信道找到对应的消费者对象，通过回调函数进行收到消息的处理*（该接口会被封装成一个任务交给线程池执行，消息处理并不在主线程运行，防止阻塞）*
9、添加响应接口：连接收到基础响应后，向通用响应队列映射中添加响应*（每次收到响应，调用此函数，把该响应放进映射）*
// 客户端的信道管理模块，一个直接面向用户的模块，内部包含多个向外提供的服务接口，用户需要什么服务，就调用客户端信道对应的服务接口
#pragma once
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../MQcommon/Log.hpp"
#include "../MQcommon/Helper.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include "../MQcommon/mq_proto.pb.h"
#include "Consumer.hpp"
#include <iostream>
#include <mutex>
#include <condition_variable>
#include <unordered_map>
using namespace std;
using namespace ns_ClientConsumer;
using namespace MQ_message;
namespace ns_client_Channel
{
typedef shared_ptr<google::protobuf::Message> MessagePtr;
using ProtobufCodecPtr = shared_ptr<ProtobufCodec>;
using BasicCommonResponsePtr = shared_ptr<BasicCommonResponse>; // 通用响应类型
using BasicConsumeResponsePtr = shared_ptr<BasicConsumeResponse>; // 推送消息响应类型
class Channel
{
public:
using ptr = shared_ptr<Channel>;
Channel(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec) : _conn(conn), _codec(codec), _Cid(UUIDHelper::Getuuid())
{
}
~Channel()
{
// 信道析构时，订阅也要对应取消
BasicCancel();
}
// 面向用户的公开接口
// 信道的创建与删除
bool OpenChannel()
{
OpenChannelRequest req;
string rid = UUIDHelper::Getuuid(); // 请求ID是唯一的
req.set_rid(rid);
req.set_cid(_Cid);
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
return bcrp->ok();
}
bool DeleteChannel()
{
CloseChannelRequest req;
string rid = UUIDHelper::Getuuid(); // 请求ID是唯一的
req.set_rid(rid);
req.set_cid(_Cid);
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
return bcrp->ok();
}
// 交换机的声明与删除
bool DeclareExchange(const string &Ename, MQ_message::ExchangeType type, bool durable, bool auto_delete, google::protobuf::Map<string, string> &qargs)
{
// 1、构建一个声明交换机的请求对象
DeclareExchangeRequest req;
string rid = UUIDHelper::Getuuid(); // 请求ID是唯一的
req.set_rid(rid);
req.set_cid(_Cid);
req.set_exchange_name(Ename);
req.set_exchange_type(type);
req.set_durable(durable);
req.set_auto_delete(auto_delete);
req.mutable_args()->swap(qargs);
// 2、向服务器发送请求
_codec->send(_conn, req);
// 3、等待服务器发来的响应
BasicCommonResponsePtr bcrp = WaitConsume(rid);
// 4、返回结果
return bcrp->ok();
}
bool DeleteExchange(const string &Ename)
{
DeleteExchangeRequest req;
string rid = UUIDHelper::Getuuid();
req.set_rid(rid);
req.set_cid(_Cid);
req.set_exchange_name(Ename);
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
return bcrp->ok();
}
// 队列的声明与删除
bool DeclareQueue(const string &Qname, bool durable, bool exclusive, bool auto_delete, google::protobuf::Map<string, string> &args)
{
DeclareQueueRequest req;
string rid = UUIDHelper::Getuuid();
req.set_rid(rid);
req.set_cid(_Cid);
req.set_queue_name(Qname);
req.set_exclusive(exclusive);
req.set_durable(durable);
req.set_auto_delete(auto_delete);
req.mutable_args()->swap(args);
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
return bcrp->ok();
}
bool DeleteQueue(const string &Qname)
{
DeleteQueueRequest req;
string rid = UUIDHelper::Getuuid();
req.set_rid(rid);
req.set_cid(_Cid);
req.set_queue_name(Qname);
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
return bcrp->ok();
}
// 交换机-队列的绑定与解绑
bool ExchangeQueueBinding(const string &Ename, const string &Qname, const string &key)
{
QueueBindRequest req;
string rid = UUIDHelper::Getuuid();
req.set_rid(rid);
req.set_cid(_Cid);
req.set_exchange_name(Ename);
req.set_queue_name(Qname);
req.set_binding_key(key);
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
return bcrp->ok();
}
bool ExchangeQueueUnBinding(const string &Ename, const string &Qname)
{
QueueUnBindRequest req;
string rid = UUIDHelper::Getuuid();
req.set_rid(rid);
req.set_cid(_Cid);
req.set_exchange_name(Ename);
req.set_queue_name(Qname);
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
return bcrp->ok();
}
// 发布消息（客户端发布消息时，不需要指定队列名称，只需要指定交换机名称，发布给交换机即可，具体放到哪个队列由交换机决定）
bool BasicPublishMessage(const string &Ename, MQ_message::BasicProperties *bp, const string &body)
{
BasicPublishRequest req;
string rid = UUIDHelper::Getuuid();
req.set_rid(rid);
req.set_cid(_Cid);
req.set_exchange_name(Ename);
req.set_body(body);
if (bp != nullptr)
{
req.mutable_properties()->set_id(bp->id());
req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
req.mutable_properties()->set_routing_key(bp->routing_key());
}
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
if (bcrp->ok())
{
lg.LogMessage(info, "消息发布成功！\n");
}
return bcrp->ok();
}
// 确认消息（确认收到哪一个队列的哪一条消息）
bool BasicAckQueueMessage(const string &msg_id)
{
BasicAckRequest req;
string rid = UUIDHelper::Getuuid();
req.set_rid(rid);
req.set_cid(_Cid);
req.set_queue_name(_consumer->_Qname); // 确认消息说明是订阅者，一定有队列名
req.set_message_id(msg_id);
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
if (bcrp->ok())
{
lg.LogMessage(info, "消息已确认\n");
}
return bcrp->ok();
}
// 队列的订阅（需要额外传入收到来自服务端的推送消息的响应后的业务处理回调函数）
bool BasicConsume(const string &consumer_tag, const string &queue_name, bool auto_ack, const CallBack &cb)
{
if (_consumer != nullptr)
{
lg.LogMessage(warning, "当前信道订阅者已订阅队列！请取消订阅后再订阅其它队列消息！\n");
return false;
}
BasicConsumeRequest req;
string rid = UUIDHelper::Getuuid();
req.set_rid(rid);
req.set_cid(_Cid);
req.set_consumer_tag(consumer_tag);
req.set_queue_name(queue_name);
req.set_auto_ack(auto_ack);
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
if (bcrp->ok() == false)
{
lg.LogMessage(errno, "订阅消息队列失败！\n");
return false;
}
_consumer = make_shared<Consumer>(consumer_tag, queue_name, auto_ack, cb); // 关键，处理消息都要依仗此函数
lg.LogMessage(info, "消费者 %s 订阅了消息队列 %s\n", consumer_tag.c_str(), queue_name.c_str());
return true;
}
// 订阅的取消（订阅后，由关联的消费者提供参数，不需要用户给予）
bool BasicCancel()
{
if (_consumer.get() == nullptr)
{
// 不是订阅者角色，不需要取消订阅，直接返回即可
return true;
}
BasicCancelRequest req;
string rid = UUIDHelper::Getuuid();
req.set_rid(rid);
req.set_cid(_Cid);
req.set_consumer_tag(_consumer->_tag);
req.set_queue_name(_consumer->_Qname);
_codec->send(_conn, req);
BasicCommonResponsePtr bcrp = WaitConsume(rid);
_consumer.reset();
return bcrp->ok();
}
// 其它一些公开接口
// 向连接提供的接口：连接收到基础响应后，向hash_map中添加响应（每次收到响应，调用此函数，把该响应放进消息队列）
void PutBasicCommonResponse(const BasicCommonResponsePtr &rep)
{
unique_lock<mutex> lock(_mutex);
_basic_resp[rep->rid()] = rep;
_cv.notify_all(); // 添加完响应后通过条件变量 (_cv) 通知所有等待的线程继续
}
// 连接收到推送消息的响应后，需要通过信道找到对应的消费者对象，通过回调函数进行消息处理（该接口会被封装成一个任务交给线程池执行，消息处理并不在主线程运行）
void Consume(const BasicConsumeResponsePtr &rep)
{
if (_consumer.get() == nullptr)
{
lg.LogMessage(error, "消息处理失败！未找到订阅者信息！\n");
return;
}
if (_consumer->_tag != rep->consumer_tag())
{
lg.LogMessage(error, "消息处理失败！收到的推送消息中的消费者标识：%s，与当前信道消费者标识：%s不一致！\n", _consumer->_tag.c_str(), rep->consumer_tag().c_str());
return;
}
_consumer->_callback(rep->consumer_tag(), rep->mutable_properties(), rep->body());
lg.LogMessage(info, "消息处理完毕\n");
}
string cid()
{
return _Cid;
}
private:
// 等待响应的函数
BasicCommonResponsePtr WaitConsume(const string &rid)
{
unique_lock<mutex> lock(_mutex);
_cv.wait(lock, [&rid, this]()
{ return _basic_resp.find(rid) != _basic_resp.end(); }); // 条件变量阻塞等待，结果完成后才继续
BasicCommonResponsePtr bcrp = _basic_resp[rid];
_basic_resp.erase(rid);
return bcrp;
}
string _Cid; // 信道ID
muduo::net::TcpConnectionPtr _conn; // 信道关联的网络通信连接对象
ProtobufCodecPtr _codec; // protobuf协议处理对象
Consumer::ptr _consumer; // 信道关联的消费者
unordered_map<string, BasicCommonResponsePtr> _basic_resp; // 请求ID对应的通用响应队列映射（只要收到的不是推送消息的响应，都放入此映射中，便于快速判断响应）
mutex _mutex; // 互斥锁
condition_variable _cv; // 条件变量，大部分请求都是阻塞操作，发送请求后进程需要阻塞，等到有响应了才能继续
};
// 信道管理类，一个连接可能有多个信道
class ChannelManager
{
public:
using ptr = shared_ptr<ChannelManager>;
ChannelManager()
{
}
// 创建信道
Channel::ptr CreateChannel(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
{
unique_lock<mutex> lock(_mutex);
auto channel = make_shared<Channel>(conn, codec);
_channels[channel->cid()] = channel;
return channel;
}
// 删除信道
void RemoveChannel(const string &Cid)
{
unique_lock<mutex> lock(_mutex);
_channels.erase(Cid);
}
// 查询信道
Channel::ptr SelectChannel(const string &Cid)
{
unique_lock<mutex> lock(_mutex);
auto it = _channels.find(Cid);
if (it == _channels.end())
{
lg.LogMessage(warning, "该信道不存在！\n");
return nullptr;
}
return it->second;
}
private:
mutex _mutex;
unordered_map<string, Channel::ptr> _channels; // Cid与信道的映射集合
};
}
2.2 通信连接模块
用户通过连接模块实例化出连接对象，连接对象创建信道，信道提供服务。因此，客户端这边的连接对用户而言，就是一个资源的载体。通信连接模块就是对客户端资源的整合。（本质上是对muduo库中TcpClient的二次封装与扩展）
2.2.1 连接管理的数据
1、muduo库中的同步控制类muduo::CountDownLatch：协调多线程之间的同步操作，确保连接的成功*（发起连接后，其他进程陷入等待；连接建立成功，其他进程结束等待）*
2、muduo库中的异步循环控制类muduo::net::EventLoopThread：为了避免网络事件和其他异步任务阻塞主线程，通过EventLoopThread创建子线程，这个线程会不断地轮询 I/O 事件来进行事件监控，该线程实例化成功会自动运行事件监控，不需要手动运行*（EventLoop实例会被绑定到TcpClient实例上，使EventLoop在监控到事件时可以调用TcpClient注册的回调函数进行处理）*
3、客户端对应的连接muduo::net::TcpConnectionPtr，当客户端连接成功后，会对其进行赋值
4、客户端muduo::net::TcpConnection类
5、 响应分发器
6、protobuf协议处理器
7、我们自己的异步工作线程
8、信道管理类
2.2.2 连接进行的操作
1、初始化各个参数
2、创建信道
3、关闭信道
4、处理推送消息的响应
// 连接管理模块（通过连接创立信道，再通过信道提供服务），通过连接搭建客户端，不需要再提供服务了
#pragma once
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"
#include "Channel.hpp"
#include "AsyncWorker.hpp"
using namespace std;
using namespace ns_client_Channel;
using namespace ns_AsyncWorker;
using namespace MQ_message;
namespace ns_client_Connection
{
class Connection
{
public:
using ptr = shared_ptr<Connection>;
// 只会收到推送消息和通用响应两种响应：BasicConsumeResponsePtr BasicCommonResponsePtr
// 传入参数：服务器IP，服务器端口号，异步工作线程
Connection(const string &Sip, int Sport, const AsyncWorker::ptr &worker) : _latch(1), // 初始化为1，说明有1个事件需要等待，内部是个计数器
_client(worker->loopthread.startLoop() /*EventLoop实例被绑定到TcpClient实例上，使EventLoop在监控到事件时可以调用TcpClient注册的回调函数进行处理*/, muduo::net::InetAddress(Sip, Sport), "Client"),
_dispatcher(bind(&Connection::onUnknownMessage, this, placeholders::_1, placeholders::_2, placeholders::_3)),
_codec(make_shared<ProtobufCodec>(bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, placeholders::_1, placeholders::_2, placeholders::_3))),
_worker(worker), _channel_manager(make_shared<ChannelManager>())
{
// 注册不同业务（推送消息、通用）响应处理函数，根据参数不同，选择不同的回调函数处理
_dispatcher.registerMessageCallback<BasicConsumeResponse>(bind(&Connection::ConsumeResponse, this, placeholders::_1, placeholders::_2, placeholders::_3));
_dispatcher.registerMessageCallback<BasicCommonResponse>(bind(&Connection::CommonResponse, this, placeholders::_1, placeholders::_2, placeholders::_3));
// 设置消息回调处理函数，对收到的数据进行协议解析（通过TcpConnection中的事件类型来判断）
_client.setMessageCallback(bind(&ProtobufCodec::onMessage, _codec, placeholders::_1, placeholders::_2, placeholders::_3));
// 建立链接成功后的回调函数（通过TcpConnection中的事件类型来判断）
_client.setConnectionCallback(bind(&Connection::onConnection, this, placeholders::_1));
// 连接服务器（非阻塞接口，即发起请求后会立刻返回，不保证连接一定成功），因此需要CountDownLatch接口，阻塞等待链接建立成功后才返回
_client.connect();
_latch.wait(); // 一直阻塞等待，直至连接建立成功（ConnectionBackFuction函数被调用）
}
// 打开/创建信道
Channel::ptr OpenChannel()
{
Channel::ptr channel = _channel_manager->CreateChannel(_conn, _codec);
bool res = channel->OpenChannel();
if (res == false)
{
lg.LogMessage(errno, "打开信道失败！\n");
return nullptr;
}
return channel;
}
// 关闭信道
void CloseChannel(const Channel::ptr &channel)
{
bool res = channel->DeleteChannel();
_channel_manager->RemoveChannel(channel->cid());
if (res == false)
{
lg.LogMessage(error, "删除信道失败！\n");
return;
}
}
private:
// 处理推送消息响应
void ConsumeResponse(const muduo::net::TcpConnectionPtr &conn, const BasicConsumeResponsePtr &message, muduo::Timestamp)
{
// 1、找到信道
Channel::ptr channel = _channel_manager->SelectChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(errno, "查询信道消息失败！\n");
return;
}
// 2、封装异步任务（消息处理任务），抛入线程池
_worker->pool.push([channel, message]()
{ channel->Consume(message); });
}
// 处理通用响应
void CommonResponse(const muduo::net::TcpConnectionPtr &conn, const BasicCommonResponsePtr &message, muduo::Timestamp)
{
// 1、找到信道
Channel::ptr channel = _channel_manager->SelectChannel(message->cid());
if (channel.get() == nullptr)
{
lg.LogMessage(errno, "查询信道消息失败！\n");
return;
}
// 2、将得到的响应对象添加到信道的通用响应hash_map中
channel->PutBasicCommonResponse(message);
}
// muduo库中没有的响应，如何处理
void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
{
LOG_INFO << "未知响应: " << message->GetTypeName() << "\n";
conn->shutdown();
}
// 建立链接成功后的回调函数，连接建立成功后，唤醒connect()的阻塞
void onConnection(const muduo::net::TcpConnectionPtr &conn)
{
if (conn->connected())
{
_latch.countDown(); // 连接成功唤醒上方阻塞
_conn = conn;
}
else
{
// 连接关闭后的操作
_conn.reset();
}
}
muduo::CountDownLatch _latch; // 协调多线程之间的同步操作，确保连接的成功（发起连接后，其他进程陷入等待；连接建立成功，其他进程结束等待）
muduo::net::EventLoopThread _loopthread; // 异步循环控制，为了避免网络事件和其他异步任务阻塞主线程，通过EventLoopThread创建子线程，这个线程会不断地轮询 I/O 事件来进行事件监控，该线程实例化成功会自动运行事件监控，不需要手动运行
muduo::net::TcpConnectionPtr _conn; // 客户端对应的连接，当客户端连接成功后，会对其进行赋值
muduo::net::TcpClient _client; // 客户端
ProtobufDispatcher _dispatcher; // 响应分发器
ProtobufCodecPtr _codec; // protobuf协议处理器
AsyncWorker::ptr _worker; // 我们自己的异步工作线程
ChannelManager::ptr _channel_manager; // 信道管理类
};
};
2.3 消费者模块
消费者子模块在客户端模块中的存在感比较低，因为用户的操作接口都是由信道提供的。一般也就是在订阅客户端的用户调用信道模块中订阅队列的操作接口的时候，与信道产生一下关联。
2.3.1 管理的消息
1、消费者标识
2、订阅的消息队列名称
3、是否自动应答标志（若为真，一个消息被消费者消费后，会直接移除该待确认信息；若为假，需要等待客户端确认）
4、消息的回调函数（客户端调用这个函数进行消息的处理）
2.3.2 管理的操作：新增消费者、删除消费者
// 订阅者模块，在客户端中并不直接向用户展示，作用只是描述当前信道是个订阅者（消费者）信道
// 项目为简化操作，一个信道只有一个消费者（实际上，不同的订阅者保存着不同的回调函数，用来处理消息）
#pragma once
#include "../MQcommon/Helper.hpp"
#include "../MQcommon/Log.hpp"
#include "../MQcommon/mq_msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <vector>
#include <mutex>
#include <memory>
#include <functional>
using namespace std;
using namespace ns_helper;
namespace ns_ClientConsumer
{
// 消费者（订阅者）结构
using CallBack = function<void(const string, const MQ_message::BasicProperties *bp, const string)>;
class Consumer
{
public:
using ptr = shared_ptr<Consumer>;
Consumer()
{
lg.LogMessage(info, "新增消费者ID：%s\n", _tag.c_str());
}
Consumer(const string &tag, const string &Qname, bool auto_ack, const CallBack &callback) : _tag(tag), _Qname(Qname), _auto_ack(auto_ack), _callback(callback)
{
lg.LogMessage(info, "新增消费者ID：%s\n", _tag.c_str());
}
~Consumer()
{
lg.LogMessage(info, "%s 消费者已退出\n", _tag.c_str());
}
string _tag; // 消费者标识
string _Qname; // 订阅的消息队列名称
bool _auto_ack; // 是否自动应答标志（若为真，一个消息被消费者消费后，会直接移除该待确认信息；若为假，需要等待客户端确认）
CallBack _callback; // 消息的回调函数（客户端调用这个函数进行消息的处理）
};
}
2.4 异步工作线程池模块
订阅者客户端在收到自服务端推送而来的消息响应后，需要对推送过来的消息进行处理，为了防止主线程阻塞影响效率，我们建立一个异步工作线程池帮助我们处理
至于为什么要把这个线程池单独封装成一个模块呢？因为这个线程池不是只服务于某个连接/信道，可以应用于多个连接多个信道，所以单独进行封装
// 异步工作线程模块，不是只服务于某个连接/信道，可以应用于多个连接多个信道，因此单独进行封装
#pragma once
#include "muduo/net/EventLoopThread.h"
#include "../MQcommon/Log.hpp"
#include "../MQcommon/Helper.hpp"
#include "../MQcommon/ThreadPool.hpp"
using namespace ns_ThreadPool;
namespace ns_AsyncWorker
{
class AsyncWorker
{
public:
using ptr = shared_ptr<AsyncWorker>;
muduo::net::EventLoopThread loopthread; // muduo库中客户端连接的异步循环线程EventLoopThread
ThreadPool pool; // 收到消息后进行异步处理的工作线程池
};
}
3、服务端、订阅者客户端、发布者客户端的main函数示例
3.1 服务端main函数Server.cpp
启动时要提前给出端口号和绑定信息数据表的存储位置
#include "Server.hpp"
int main()
{
ns_Server::Server server(8085, "./data/");
server.start();
return 0;
}
3.2 订阅者客户端main函数ConsumeClient.cpp
// 订阅消息的消费者（订阅者）客户端
#include "Connection.hpp"
using namespace std;
using namespace ns_AsyncWorker;
using namespace ns_client_Channel;
using namespace ns_client_Connection;
void cb(Channel::ptr &channel, const string consumer_tag, const BasicProperties *bp, const string &body)
{
lg.LogMessage(info, "正在进行消息处理……\n");
lg.LogMessage(info, "%s 成功订阅了消息：%s\n", consumer_tag.c_str(), body.c_str());
channel->BasicAckQueueMessage(bp->id()); // 确认消息
}
// 回调函数
int main(int argc, char *argv[])
{
if (argc != 2)
{
lg.LogMessage(error, "输入错误，你应该这么输入：./ConsumeClient + 欲订阅的消息队列名\n");
return -1;
}
// 1、实例化异步工作线程对象
AsyncWorker::ptr worker = make_shared<AsyncWorker>();
// 2、实例化连接对象
Connection::ptr conn = make_shared<Connection>("127.0.0.1", 8085, worker);
// 3、通过连接创建信道
Channel::ptr channel = conn->OpenChannel();
// 4、通过信道提供的服务完成所需
// 4.1 声明一个交换机Exchange1，交换机类型为广播模式
google::protobuf::Map<std::string, std::string> tmp_map;
channel->DeclareExchange("exchange1", MQ_message::ExchangeType::topic, true, false, tmp_map);
// 4.2 声明一个队列Queue1
channel->DeclareQueue("queue1", true, false, false, tmp_map);
// 4.3 声明一个队列Queue2
channel->DeclareQueue("queue2", true, false, false, tmp_map);
// 4.4 绑定queue1-exchange1，且binding_key设置为queue1
channel->ExchangeQueueBinding("exchange1", "queue1", "queue1");
// 4.5 绑定queue2-exchange1，且binding_key设置为news.music.#
channel->ExchangeQueueBinding("exchange1", "queue2", "news.music.#");
// 5、订阅消息
auto functor = std::bind(cb, channel, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
channel->BasicConsume("consumer1", argv[1], false, functor);
while (1)
{
std::this_thread::sleep_for(std::chrono::seconds(3)); // 每三秒执行一次
}
conn->CloseChannel(channel);
return 0;
}
3.3 发布者客户端PublishClient.cpp
// 发布消息的生产者（发布者）客户端
#include "Connection.hpp"
using namespace std;
using namespace ns_AsyncWorker;
using namespace ns_client_Channel;
using namespace ns_client_Connection;
int main()
{
// 1、实例化异步工作线程对象
AsyncWorker::ptr worker = make_shared<AsyncWorker>();
// 2、实例化连接对象
Connection::ptr conn = make_shared<Connection>("127.0.0.1", 8085, worker);
// 3、通过连接创建信道
Channel::ptr channel = conn->OpenChannel();
// 4、通过信道提供的服务完成所需
// 4.1 声明一个交换机Exchange1，声明交换机类型
google::protobuf::Map<std::string, std::string> tmp_map;
// 广播模式测试 channel->DeclareExchange("exchange1", MQ_message::ExchangeType::fanout, true, false, tmp_map);
// 直接交换模式测试 channel->DeclareExchange("exchange1", MQ_message::ExchangeType::direct, true, false, tmp_map);
channel->DeclareExchange("exchange1", MQ_message::ExchangeType::topic, true, false, tmp_map);
// 4.2 声明一个队列Queue1
channel->DeclareQueue("queue1", true, false, false, tmp_map);
// 4.3 声明一个队列Queue2
channel->DeclareQueue("queue2", true, false, false, tmp_map);
// 4.4 绑定queue1-exchange1，且binding_key设置为queue1
channel->ExchangeQueueBinding("exchange1", "queue1", "queue1");
// 4.5 绑定queue2-exchange1，且binding_key设置为news.music.#
channel->ExchangeQueueBinding("exchange1", "queue2", "news.music.#");
// 5、循环向交换机发布消息
for (int i = 0; i < 1; i++)
{
/*
广播模式测试
channel->BasicPublishMessage("exchange1", nullptr, "Hello World-" + to_string(i));
*/
BasicProperties bp;
bp.set_id(UUIDHelper::Getuuid());
bp.set_delivery_mode(DeliveryMode::durable);
bp.set_routing_key("news.music.pop");
// 直接交换模式测试 bp.set_routing_key("queue1");
channel->BasicPublishMessage("exchange1", &bp, "Hello World-" + std::to_string(i));
}
BasicProperties bp;
bp.set_id(UUIDHelper::Getuuid());
bp.set_delivery_mode(DeliveryMode::durable);
bp.set_routing_key("news.music.sport");
channel->BasicPublishMessage("exchange1", &bp, "这是动感音乐新闻！");
bp.set_routing_key("news.sport");
channel->BasicPublishMessage("exchange1", &bp, "这是运动新闻！");
// 6、关闭信道
conn->CloseChannel(channel);
return 0;
}
本文转载自: https://blog.csdn.net/m0_73682725/article/details/141289079
版权归原作者 东方未明0108 所有， 如有侵权，请联系我们删除。
版权归原作者 东方未明0108 所有， 如有侵权，请联系我们删除。