当前位置：首页 > news >正文

【RabbitMQ 项目】客户端：连接模块

news 2025/5/8 23:49:16

文章目录

一.实现要点
- 构造函数
二.代码实践
三.搭建消费客户端和生产客户端

客户端有两种，生产客户端，消费客户端，其实连接模块就是传统意义上的的客户端，生产客户端，消费客户端都是用它来搭建的，只不过连接模块提供了很多接口。消费客户端只会使用其中的一部分，比如消费客户端会发布消息；生产客户端也只使用一部分，比如不可能订阅消息

一.实现要点

构造函数

muduo 库中的 TcpClient 成员的注册两个回调函数，一个是从接收缓冲区读取上来消息后的回调函数，注册为协议处理器的 onMessage 成员，另一个是连接建立成功后的回调，我们自己实现一个，工作就是给 Connection 内部维护的 TCP 连接赋值为这个新建的连接
此外，为分发器注册业务处理函数，即对不同响应的处理动作。第一，收到一个 CommonResponse，结合我们上节介绍的信道模块，则需要向信道的哈希表中 put 该 Response；第二，收到一个 PushMessageResponse，我们需要找到信道，调用信道的提供的处理消息的函数(工作是先找到消费者，然后调用消费者的回调函数)。但这个任务属于支线任务，Reactor 线程不想自己做，于是交给线程池来做
最后，调用 TcpClient 的 connect 接口，向服务端发起连接

二.代码实践

#pragma once
#include "muduo/protobuf/codec.h"
#include "muduo/protobuf/dispatcher.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 "../common/ThreadPool.hpp"
#include "Channel.hpp"
#include <functional>
#include <iostream>
namespace ns_connection
{using ThreadPoolPtr = std::shared_ptr<ns_tp::ThreadPool>;using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;using ProtobufDispatcherPtr = std::shared_ptr<ProtobufDispatcher>;using ChannelPtr = std::shared_ptr<ns_channel::Channel>;using ThreadPoolPtr = std::shared_ptr<ns_tp::ThreadPool>;using CommonResponsePtr = std::shared_ptr<ns_protocol::CommomResponse>;using PushMessageResonsePtr = std::shared_ptr<ns_protocol::PushMessageResponse>;/************ Connection是对底层用于通信的TCP套接字封装(muduo库中的TcpConnectionPtr)* 一个Connection中包含多个信道，当Connection关闭，信道也会销毁* ******************/class Connection{private:muduo::net::EventLoopThread _loopThread;muduo::CountDownLatch _latch;muduo::net::TcpClient _client;muduo::net::TcpConnectionPtr _connPtr;ProtobufDispatcherPtr _distpatcherPtr;ProtobufCodecPtr _codecPtr;ns_channel::ChannelManager _channelManager;ThreadPoolPtr _threadPoolPtr;public:Connection(const std::string &serverIp, int serverPort, const ThreadPoolPtr &threadPoolPtr): _loopThread(),_latch(1),_client(_loopThread.startLoop(), muduo::net::InetAddress(serverIp, serverPort), "client"),_connPtr(),_channelManager(),_threadPoolPtr(threadPoolPtr){// 构造成员_distpatcherPtr = std::make_shared<ProtobufDispatcher>((std::bind(&Connection::onUnknownMessage,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3)));_codecPtr = std::make_shared<ProtobufCodec>((std::bind(&ProtobufDispatcher::onProtobufMessage,_distpatcherPtr.get(),std::placeholders::_1,std::placeholders::_2,std::placeholders::_3)));// 给Client注册两个回调函数_client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));_client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codecPtr.get(), std::placeholders::_1,std::placeholders::_2, std::placeholders::_3));_distpatcherPtr->registerMessageCallback<ns_protocol::CommomResponse>(std::bind(&Connection::onCommonResponse,this, std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));_distpatcherPtr->registerMessageCallback<ns_protocol::PushMessageResponse>(std::bind(&Connection::onRecvMessage,this, std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));connect();}void connect(){_client.connect(); // 非阻塞_latch.wait();}ChannelPtr openChannel(){// 只是在本地建立了信道auto channelPtr = _channelManager.openChannel(_connPtr, _codecPtr);// 通过该信道发送建立信道的请求，要服务端也建立对应的信道if (!channelPtr->openChannel()){LOG(WARNING) << "打开信道失败" << endl;// 关闭本地的信道，防止内存泄漏_channelManager.closeChannel(channelPtr->_id);}return channelPtr;}void closeChannel(const ChannelPtr& channelPtr){// 发送关闭信道的请求，让服务端关闭信道channelPtr->closeChannel();// 把本地信道关掉_channelManager.closeChannel(channelPtr->_id);}private:// 给_client设置的回调void onConnection(muduo::net::TcpConnectionPtr connPtr){if (connPtr->connected()){_connPtr = connPtr;_latch.countDown();}else{_connPtr.reset();}}void onUnknownMessage(const muduo::net::TcpConnectionPtr &connPtr,const MessagePtr &resp, muduo::Timestamp time){LOG(WARNING) << "未知响应" << endl;}// 业务处理函数void onCommonResponse(const muduo::net::TcpConnectionPtr &connPtr,const CommonResponsePtr &respPtr, muduo::Timestamp time){//LOG(DEBUG) << "收到CommonResponse, respId: " << respPtr->response_id() << endl;std::string channeId = respPtr->channel_id();auto channelPtr = _channelManager.getChannel(channeId);channelPtr->putCommonResponse(respPtr);}void onRecvMessage(const muduo::net::TcpConnectionPtr &connPtr,const PushMessageResonsePtr &respPtr, muduo::Timestamp time){//LOG(DEBUG) << "收到消息, body: " << respPtr->msg().saved_info().body() << endl;std::string channeId = respPtr->channel_id();auto channelPtr = _channelManager.getChannel(channeId);// 把处理消息的任务交给线程池来做_threadPoolPtr->push(std::bind(&ns_channel::Channel::consumeMessage, channelPtr.get(),respPtr->qname(), respPtr->msg()));}};
}