仿RabbitMQ实现消息队列（三）
muduo库
muduo 是一个基于 C++11 的高性能异步网络库，由陈硕开发，采用非阻塞 I/O 模型和事件驱动架构，专为 Linux 平台设计，采用 Reactor 模式实现高并发网络编程。它不仅提供了核心的网络通信组件，还包含了日志、线程、内存管理等基础工具库，形成了一套完整的服务端开发框架。它其实是一款基于主从Reactor模型的网络库，其使用的线程模型是one loop per thread,所以one loop per thread指的是：
- 一个线程只能有一个事件循环(EventLoop),用于响应计时器和io事件，用于响应计时器和IO事件
- 一个文件描述符只能由一个线程进行读写，也就是一个TCP连接必须归属于某个EventLoop管理
Reactor模式
Reactor模型是一种事件驱动的设计模式，主要用于处理高并发的I/O操作，尤其是适用于需要高校处理大量并发连接的场景（如网络服务器，分布式系统等）。核心思想是通过一个或多个“反应器（Reactor）”来统一管理和分发事件，避免传统多线程模型中线程创建/切换的开销，从而提升系统的并发处理能力。
库介绍
EventLoop.h
1 | class EventLoop : noncopyable |
在本次项目中我们只需要直到loop是启动事件监控即可。和服务器一起启动。
TcpServer类基础介绍
这个类主要职责是：
1、绑定并监听指定端口（通过内部的Acceptor组件完成）。
2、当有新客户端连接到来时，自动创建TcpConnection对象管理该连接。
3、向用户提供回调接口，用于处理连接建立、消息接收、连接管理关闭等事件。
4、1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25typedef 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, const string& nameArg, Option option = kNoReusePort);
void setThreadNum(int numThreads);
void start();
/// 当一个新连接建立成功的时候被调用
void setConnectionCallback(const ConnectionCallback& cb)
{ connectionCallback_ = cb; }
/// 消息的业务处理回调函数---这是收到新连接消息的时候被调用的函数
void setMessageCallback(const MessageCallback& cb) { messageCallback_ = cb; }
};
说明：
- 构造函数：
- loop: 进行事件监控的对象。
- listenAddr: 绑定的ip,port
- nameArg:服务器的名称
- option:套接字选项。是否启用端口复用，默认不启用*
- setThreadNum:设置线程数目。
- start: 启动服务器。
- setConnectionCallback:设置收到新链接的时候要调用的回调函数。
- setMessageCallback: 设置收到客户端发来的信息时要调用的回调函数，也就是处理数据的函数。*
TcpConnection
在此项目中，我们只需要以下接口：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33class 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); // this one will swap data void shutdown();
// NOT thread safe, no simultaneous calling
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_;
};
connected：判断链接的状态，即是否是链接状态。
send:向对端发送数据。TcpClient类
1 | class TcpClient : noncopyable |
说明：
- 构造函数：
- loop:事件监控对象。
- serverAddr:服务器地址。
- nameArg:客户端名称。
- connected():连接服务器。
Buffer类
了解retrieveAllAsString()接口，就是将数据转换为字符串muduo快速上手
搭建个简单的服务器
服务端1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
using namespace muduo;
using namespace muduo::net;
class TranslateServer
{
public:
TranslateServer(int port):_server(&_baseloop,muduo::net::InetAddress("0.0.0.0",port),
"translate",muduo::net::TcpServer::kReusePort)
{
//将我们的类成员函数，处理成服务器的回调处理函数 _server.setConnectionCallback(std::bind(&TranslateServer::onConnection,this,std::placeholders::_1));
_server.setMessageCallback(std::bind(&TranslateServer::onMessage,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
}
//启动服务器 事件开始后 收到消息后会自动调用回调函数进行业务处理
void start(){
//都是开始事件监听
_server.start();
_baseloop.loop();//这是个死循环阻塞接口
}
void onConnection(const TcpConnectionPtr& conn)
{
if(conn->connected()==true)
{
std::cout<<"新建成功"<<std::endl;
}
else{
std::cout<<"连接失败"<<std::endl;
}
}
std::string translate(const std::string& str)
{
std::unordered_map<std::string,std::string>dict_map={
{"hello","你好"},
{"吃饭没","eat"},
{"火鸡面","tickoo"},
{"吃了","油泼面"}
};
auto it=dict_map.find(str);
if(it==dict_map.end())
{
std::cout<<"没听懂"<<std::endl;
}
return it->second;
}
void onMessage(const muduo::net::TcpConnectionPtr& conn,Buffer* buf,muduo::Timestamp)
{
//通信连接收到请求时的回调函数
//从buf中把数据取出来
std::string str=buf->retrieveAllAsString();
//调用translate进行翻译
std::string resp=translate(str);
//对客户端进行相应
conn->send(resp);
}
private:
//上面参数要传到下面这个参数里面
//baseloop是epoll的事件监控，会进行描述符的事件监控，触发事件后进行io操作
EventLoop _baseloop;
//这个server主要用于设置回调函数，告诉服务器收到什么请求该如何处理
muduo::net::TcpServer _server;
};
int main()
{
TranslateServer ser(8085);
ser.start();
return 0;
}
整个业务逻辑非常简单，先设置回调处理，开始事件监控，会自动收到消息后调用回调。
编译1
2server:dict_server.cpp
g++ -o $@ $^ -I../lib/build/release-install-cpp11/include -L../lib/build/release-install-cpp11/lib -lmuduo_net -lmuduo_base -pthread
由于我的cpp文件和muduo库这个文件同属于一个文件内，并且我这个安装路径在 lib/ 目录下，而我的lib目录
用下面指令查看我们的端口号是否被监听1
netstat -anptu | grep 8085
客户端：1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
using namespace muduo;
using namespace muduo::net;
class TranslateClient
{
public:
TranslateClient(const std::string& ip,int port):_latch(1),_client(_loopthread.startLoop(),
muduo::net::InetAddress(ip,port),"TranslateClient"){
_client.setConnectionCallback(std::bind(&TranslateClient::onConnection,this,std::placeholders::_1));
_client.setMessageCallback(std::bind(&TranslateClient::onMessage,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
}
void connect()
{
_client.connect();
//阻塞等待
_latch.wait();
}
bool send(const std::string msg)
{
if(_conn->connected())
_conn->send(msg);
return true;
return false;
}
private:
//连接建立成功的回调函数
void onConnection(const TcpConnectionPtr& conn){
if(conn->connected())
{
//唤醒主线程中的阻塞
_latch.countDown();
_conn=conn;
}
else{
//连接关闭
_conn.reset();
}
}
//收到消息时的回调函数
void onMessage(const TcpConnectionPtr& conn,Buffer* buf,Timestamp){
std::cout<<"翻译结果为："<<buf->retrieveAllAsString()<<std::endl;
}
private:
CountDownLatch _latch;
EventLoopThread _loopthread;
TcpClient _client;
TcpConnectionPtr _conn; //
};
int main()
{
TranslateClient client("1227.0.0.1",8085);
client.connect();
while(1)
{
std::string msg;
std::cin>>msg;
client.send(msg);
}
return 0;
}
说明：
- 成员属性：
- _latch：封装了条件变量的对象，用来确保正确获得连接，必须初始化为1。
- _loopthread：客户端不能像服务器一样一直进行事件监控的循环
- EventLoopThread对象就可以重新创建一个线程来进行事件监控，必须在_client之前初始化。
- _client：客户端对象。
- _conn：建立好的连接，用来发送数据。
- connect()：关键要等待连接真正建立好，将_conn初始化。
- send()：向服务器发送数据。
- onConnection()：关键在于建立连接后初始化_conn，并唤醒客户端。
- onMessage()：收到服务器应答后调用的回调函数
没有协议问题，还会出现tcp粘包问题，
