RabbitMQ 客户端连接管理与信道搭建实践

1. 连接管理

在客户端这边，RabbitMQ 弱化了客户端的概念，因为用户所需的服务都是通过信道来提供的，因此操作思想转换为先创建连接，通过连接创建信道，通过信道提供服务这一流程。

这个模块同样是针对 muduo 库客户端连接的二次封装，向用户提供创建 channel 信道的接口，创建信道后，可以通过信道来获取指定服务。

#ifndef __M_CONNECTION_H__
#define __M_CONNECTION_H__
#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 "worker.hpp"

namespace rabbitmq {
class Connection {
public:
    using ptr = std::shared_ptr<Connection>;
    Connection(const std::string server_ip, int server_port, const AsyncWorker::ptr& worker)
        : _latch(1),
          _client(worker->_loopthread.startLoop(), muduo::net::InetAddress(server_ip, server_port), ),
          _dispatcher(std::(&Connection::onUnknownMessage, , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
          _codec(std::<ProtobufCodec>(std::(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
          _worker(worker),
          _channel_manager(std::<ChannelManager>()) {
        _dispatcher.<rabbitmq::basicCommonResponse>(std::(&Connection::basicResponse, , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _dispatcher.<rabbitmq::basicConsumeResponse>(std::(&Connection::consumeResponse, , std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _client.(std::(&ProtobufCodec::onMessage, _codec.(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        _client.(std::(&Connection::onConnection, , std::placeholders::_1));
        _client.();
        _latch.(); 
    }

    {
        Channel::ptr channel = _channel_manager->(_conn, _codec);
         ret = channel->();
         (ret == ) {
            ();
             Channel::();
        }
         channel;
    }

    {
        channel->();
        _channel_manager->(channel->());
    }

:
    {
        
        Channel::ptr channel = _channel_manager->(message->());
         (channel.() == ) {
            ();
            ;
        }
        
        channel->(message);
    }

    {
        
        Channel::ptr channel = _channel_manager->(message->());
         (channel.() == ) {
            ();
            ;
        }
        
        _worker->_pool.([channel, message]() { channel->(message); });
    }

    {
        LOG_INFO <<  << message->();
        conn->();
    }

    
    {
         (conn->()) {
            _latch.(); 
            _conn = conn;
        }  {
            
            _conn.();
        }
    }

:
    muduo::CountDownLatch _latch; 
    muduo::net::TcpConnectionPtr _conn; 
    muduo::net::TcpClient _client; 
    ProtobufDispatcher _dispatcher; 
    ProtobufCodecPtr _codec; 
    AsyncWorker::ptr _worker;
    ChannelManager::ptr _channel_manager;
};
}