基于装饰者模式的thrift连接池实现(C++版本)

就一个连接池而已，为什么要采用装饰者模式，听着咋感觉那么高深？

所谓连接池，也就是连接先不close，放入池中，等待下次需要用的时候，直接从池中取出即可用，省去了tcp握手的时间，所以这个可以看作是一个长连接，知道连接池清除空闲连接把多余的连接清除掉才会释放。

而普通的连接，用完后如果生命周期内不再使用了，就会销毁掉。

所以基于普通的连接，要定义出线程池的那种用完放回线程池的连接，就需要把close的方法进行重写，所以可以采用继承的方法实现，相当于产生了两个子类，一个是普通连接类，一个是线程池化的连接类。

如图，如果我还想有个连接，它是优化了传输性能处理的，岂不是还要实现一个具体类？那如果我需要一个线程池化以及优化性能的连接呢？再生成一个类吗？显然不太可取了，类型会变得非常多。

这个时候可以使用装饰者模式来实现。

ThriftTransportPool::init() // 简化实现
{
    std::vector<std::shared_ptr<ITransportWrapper>> pool;
    pool[i] = std::make_shared<PooledTransport>(std::make_shared<TransportWrapper>(transport), \
        this->shared_from_this()); // 将普通连接进行装饰
}

ServerInfo serverInfo("127.0.0.1", 8080);
ThriftTransportPool transportPool(5, serverInfo); // 5个连接

std::shared_ptr<ITransportWrapper> transportWrapper = transportPool.get(); // 从池中获取
std::shared_ptr<TProtocol> protocol(new TBinaryProtocol(transportWrapper->getTransport()));
CalculatorClient client(protocol);
client.ping();
transportWrapper.close(); // 放入连接池中

std::make_shared<OptiTransport>(std::make_shared<PooledTransport>(std::make_shared<TransportWrapper>(transport), \
    this->shared_from_this())) //如果想要加入性能优化功能，再装饰一下就行

如果想要加入性能优化功能，再使用OptiTransport装饰一下就行，所以后期扩展比较方便，不会产生类爆炸的问题，新加特性增强，只需要无限套娃即可。

源码实现

包装下服务器相关信息。

// ServerInfo.hpp
#pragma once

#include <iostream>
#include <string>

using namespace std;

class ServerInfo
{
public:
    ServerInfo(string host, int port)
    {
        this->host = host;
        this->port = port;
    }

    string getHost()
    {
        return host;
    }

    void setHost(string host)
    {
        this->host = host;
    }

    int getPort()
    {
        return port;
    }

    void setPort(int port)
    {
        this->port = port;
    }

    string tostring()
    {
        return "host:" + host + ",port:" + to_string(port);
    }

private:
    string host;
    int port;
};

包装下thrift中的transport，类似于http中的socket。该包装类TransportWrapper可以查看和控制transport的一些生存状态。

// ITransportWrapper.h
#pragma once

#include <iostream>
#include <string>
#include <ctime>

#include <thrift/transport/TTransportUtils.h>

using namespace std;
using namespace apache::thrift::transport;

class ITransportWrapper
{
public:
    typedef std::shared_ptr<TTransport> TTransportPtr;
    ITransportWrapper() = default;
    virtual ~ITransportWrapper()
    {}

    virtual bool isBusy() = 0;
    virtual void setIsBusy(bool isBusy) = 0;
    virtual bool isDead() = 0;
    virtual bool isOpen() = 0;
    virtual void setIsDead(bool isDead) = 0;
    virtual TTransportPtr getTransport() = 0;
    virtual void setTransport(TTransportPtr transport) = 0;
    /**
     * 当前transport是否可用
     *
     * @return
     */
    virtual bool isAvailable() = 0;
    virtual time_t getLastUseTime() = 0;
    virtual void setLastUseTime(time_t lastUseTime) = 0;
    virtual string getHost() = 0;
    virtual int getPort() = 0;
    virtual string tostring() = 0;

    virtual void open() = 0;
    virtual void destroy() = 0;
    virtual void close() = 0;
};

// TransportWrapper.hpp
#pragma once

#include "ITransportWrapper.h"

#include <iostream>
#include <string>
#include <ctime>

#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TThreadPoolServer.h>
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TTransportUtils.h>

using namespace std;
using namespace apache::thrift;
using namespace apache::thrift::concurrency;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;

class TransportWrapper : public ITransportWrapper
{
public:
    typedef std::shared_ptr<TTransport> TTransportPtr;
    typedef std::shared_ptr<TSocket> TSocketPtr;
    TransportWrapper(string host, int port, bool isOpen = false)
    {
        TSocketPtr socket(new TSocket(host, port));
        TTransportPtr transport(new TFramedTransport (socket));
        this->lastUseTime = time(0);
        this->transport = transport;
        this->host = host;
        this->port = port;
        if (isOpen)
        {
            try
            {
                transport->open();
            }
            catch (TException &e)
            {
                std::cout << host << ":" << port << " " << e.what() << std::endl;
                isDead_ = true;
            }
        }
    }

    TransportWrapper() = default;

    ~TransportWrapper() = default;

    bool isBusy() override
    {
        return isBusy_;
    }

    void setIsBusy(bool isBusy_) override
    {
        this->isBusy_ = isBusy_;
    }

    bool isDead() override
    {
        return isDead_;
    }

    bool isOpen() override
    {
        return transport->isOpen();
    }

    void setIsDead(bool isDead_) override
    {
        this->isDead_ = isDead_;
    }

    TTransportPtr getTransport() override
    {
        return transport;
    }

    void setTransport(TTransportPtr transport) override
    {
        this->transport = transport;
    }

    /**
     * 当前transport是否可用
     *
     * @return
     */
    bool isAvailable() override
    {
        return !isBusy_ && !isDead_ && transport->isOpen();
    }

    time_t getLastUseTime() override
    {
        return lastUseTime;
    }

    void setLastUseTime(time_t lastUseTime) override
    {
        this->lastUseTime = lastUseTime;
    }

    string getHost() override
    {
        return host;
    }

    int getPort() override
    {
        return port;
    }

    string tostring() override
    {
        return host + ":" + to_string(port) + ",isBusy:" + to_string(isBusy_) + ",isDead:" + to_string(isDead_) + ",isOpen:" +
               to_string(transport->isOpen()) + ",isAvailable:" + to_string(isAvailable()) + ",lastUseTime:" + ctime(&lastUseTime);
    }

    void open() override
    {
        if (!transport->isOpen())
        {
            try
            {
                transport->open();
            }
            catch (TException &e)
            {
                std::cout << host << ":" << port << " " << e.what() << std::endl;
                isDead_ = true;
            }
        }
    }

    /**
     * 默认关闭连接
     *
     * @return
     */
    void destroy() override
    {
        this->transport->close();
    }

    /**
     * 默认关闭连接
     *
     * @return
     */
    void close() override
    {
        this->transport->close();
    }

private:
    TTransportPtr transport;

    /**
     * 是否正忙
     */
    bool isBusy_ = false;

    /**
     * 是否已经挂
     */
    bool isDead_ = false;

    /**
     * 最后使用时间
     */
    time_t lastUseTime;

    /**
     * 服务端Server主机名或IP
     */
    string host;

    /**
     * 服务端Port
     */
    int port;
};

上面的TransportWrapper类为ITransportWrapper类的普通实现，transport用完后就会释放连接，而对于连接池用到的transport，要求用完后需要放回到连接池中，而不是释放连接，否则tcp就要不停的握手了。所以下面用装饰模式来实现能自动放回连接池的transport包装类，即PoolingTransport。

// ITransportDecorator.h
#pragma once

#include "ITransportWrapper.h"

#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/server/TThreadPoolServer.h>
#include <thrift/transport/TSocket.h>
#include <thrift/concurrency/ThreadManager.h>
#include <thrift/transport/TTransportUtils.h>

using namespace std;
using namespace apache::thrift;
using namespace apache::thrift::concurrency;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;
using namespace apache::thrift::server;

class ITransportDecorator : public ITransportWrapper
{
public:
    typedef std::shared_ptr<TTransport> TTransportPtr;
    ITransportDecorator(std::shared_ptr<ITransportWrapper> _transportWrapper) : transportWrapper(_transportWrapper)
    {
    }

    ITransportDecorator() = default;

    ~ITransportDecorator()
    {}

    virtual bool isBusy() override
    {
        return transportWrapper->isBusy();
    }

    virtual void setIsBusy(bool isBusy) override
    {
        transportWrapper->setIsBusy(isBusy);
    }

    virtual bool isDead() override
    {
        return transportWrapper->isDead();
    }

    virtual bool isOpen() override
    {
        return transportWrapper->isOpen();
    }

    virtual void setIsDead(bool isDead) override
    {
        transportWrapper->setIsDead(isDead);
    }

    virtual TTransportPtr getTransport() override
    {
        return transportWrapper->getTransport();
    }

    virtual void setTransport(TTransportPtr transport) override
    {
        transportWrapper->setTransport(transport);
    }

    /**
     * 当前transport是否可用
     *
     * @return
     */
    virtual bool isAvailable() override
    {
        return transportWrapper->isAvailable();
    }

    virtual time_t getLastUseTime() override
    {
        return transportWrapper->getLastUseTime();
    }

    virtual void setLastUseTime(time_t lastUseTime) override
    {
        transportWrapper->setLastUseTime(lastUseTime);
    }

    virtual string getHost() override
    {
        return transportWrapper->getHost();
    }

    virtual int getPort() override
    {
        return transportWrapper->getPort();
    }

    virtual string tostring() override
    {
        return transportWrapper->tostring();
    }

    virtual void open() override
    {
        transportWrapper->open();
    }

    /**
     * 默认关闭连接
     *
     * @return
     */
    virtual void destroy() override
    {
        transportWrapper->destroy();
    }

    /**
     * 默认关闭连接，可被装饰成别的功能，比如放回连接池中
     *
     * @return
     */
    virtual void close() override
    {
        transportWrapper->close();
    }

protected:
    std::shared_ptr<ITransportWrapper> transportWrapper;
};

// PoolingTransport.hpp
#pragma once

#include "ITransportDecorator.h"
#include "ITransportWrapper.h"
#include "TransportWrapper.hpp"
#include "ITransportPool.h"

#include <iostream>
#include <memory>

class PoolingTransport : public ITransportDecorator, public std::enable_shared_from_this<PoolingTransport>
{
public:
    PoolingTransport(std::shared_ptr<ITransportWrapper> _transportWrapper, std::shared_ptr<ITransportPool> _transportPool)
        : ITransportDecorator(_transportWrapper), transportPool(_transportPool)
    {
    }

    PoolingTransport() = default;

    ~PoolingTransport() = default;

    /**
     * 默认关闭连接，装饰成放回连接池中
     *
     * @return
     */
    virtual void close() override
    {
        std::shared_ptr<ITransportPool> pool = transportPool.lock();
        if (pool)
        {
            transportPool->release(this->shared_from_this());
        }
    }

private:
    std::weak_ptr<ITransportPool> transportPool;
};

至于这个装饰类实现后怎么用，然后是重头戏，连接池类，可以从该连接池中取得transport的包装类，用于通信，用完后自动释放到连接池中，这里就要对普通的类再调用装饰类进行装饰，初始化连接池连接时，就是包装好的能自动放进连接池的transportWrapper。

// ITransportPool.h
#pragma once

#include <iostream>
#include <string>
#include <memory>

#include "ITransportWrapper.h"

using namespace std;
using namespace apache::thrift::transport;

class ITransportPool
{
public:
    typedef std::shared_ptr<ITransportWrapper> ITransportWrapperPtr;

    ITransportPool()
    {}
    virtual ~ITransportPool()
    {}
    virtual int getCheckInvervalSecond() = 0;
    virtual void setCheckInvervalSecond(int checkInvervalSecond) = 0;
    /**
     * 从池中取一个可用连接
     * @return
     */
    virtual ITransportWrapperPtr get() = 0;
    /**
     * 客户端调用完成后，必须手动调用此方法，将TTransport恢复为可用状态
     *
     * @param client
     */
    virtual void release(ITransportWrapperPtr client) = 0;
    virtual void destory() = 0;
    /**
     * 获取当前已经激活的连接数
     *
     * @return
     */
    virtual int getActiveCount() = 0;
    /**
     * 获取当前繁忙状态的连接数
     *
     * @return
     */
    virtual int getBusyCount() = 0;
    /**
     * 获取当前已"挂"掉连接数
     *
     * @return
     */
    virtual int getDeadCount() = 0;
    virtual string tostring() = 0;
    virtual string getWrapperInfo(ITransportWrapperPtr client) = 0;

    virtual void init() = 0;
};

// ThriftTransportPool.hpp
#pragma once

#include "ITransportPool.h"
#include "TransportWrapper.hpp"
#include "ServerInfo.hpp"
#include "PoolingTransport.hpp"

#include <iostream>
#include <string>
#include <thread>
#include <vector>
#include <memory>
#include <exception>
#include <mutex>

#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TTransportUtils.h>
#include <semaphore.h>

using namespace std;
using namespace apache::thrift;
using namespace apache::thrift::concurrency;
using namespace apache::thrift::protocol;
using namespace apache::thrift::transport;

class ThriftTransportPool : public ITransportPool, public std::enable_shared_from_this<ThriftTransportPool>
{
public:
    typedef std::shared_ptr<ITransportWrapper> ITransportWrapperPtr;
    sem_t access;
    std::mutex mt;
    std::vector<ITransportWrapperPtr> pool;
    int poolSize = 1;             //连接池大小
    int minSize = 1;              //池中保持激活状态的最少连接个数
    int maxIdleSecond = 300;      //最大空闲时间（秒），超过该时间的空闲时间的连接将被关闭
    int checkInvervalSecond = 60; //每隔多少秒，检测一次空闲连接（默认60秒）
    std::vector<ServerInfo> serverInfos;
    bool allowCheck = true;
    std::thread checkThread;

    /**
     * 连接池构造函数，多台服务器
     *
     * @param poolSize            连接池大小
     * @param minSize             池中保持激活的最少连接数
     * @param maxIdleSecond       单个连接最大空闲时间，超过此值的连接将被断开
     * @param checkInvervalSecond 每隔多少秒检查一次空闲连接
     * @param serverList          服务器列表
     */
    ThriftTransportPool(int poolSize, int minSize, int maxIdleSecond, int checkInvervalSecond, vector<ServerInfo> &serverList)
    {
        construct(poolSize, minSize, maxIdleSecond, checkInvervalSecond, serverList);
    }

    /**
     * 连接池构造函数，单台服务器
     *
     * @param poolSize            连接池大小
     * @param minSize             池中保持激活的最少连接数
     * @param maxIdleSecond       单个连接最大空闲时间，超过此值的连接将被断开
     * @param checkInvervalSecond 每隔多少秒检查一次空闲连接
     * @param serverList          服务器列表
     */
    ThriftTransportPool(int poolSize, int minSize, int maxIdleSecond, int checkInvervalSecond, ServerInfo &server)
    {
        std::vector<ServerInfo> serverList;
        serverList.push_back(server);
        construct(poolSize, minSize, maxIdleSecond, checkInvervalSecond, serverList);
    }

    /**
     * 连接池构造函数（默认最大空闲时间300秒），多台服务器
     *
     * @param poolSize   连接池大小
     * @param minSize    池中保持激活的最少连接数
     * @param serverList 服务器列表
     */
    ThriftTransportPool(int poolSize, int minSize, vector<ServerInfo> &serverList) : ThriftTransportPool(poolSize, minSize, 300, 60, serverList)
    {
    }

    ThriftTransportPool(int poolSize, vector<ServerInfo> &serverList) : ThriftTransportPool(poolSize, 1, 300, 60, serverList)
    {
    }

    ThriftTransportPool(vector<ServerInfo> &serverList) : ThriftTransportPool(serverList.size(), 1, 300, 60, serverList)
    {
    }

    /**
     * 连接池构造函数（默认最大空闲时间300秒），单台服务器
     *
     * @param poolSize   连接池大小
     * @param minSize    池中保持激活的最少连接数
     * @param serverList 服务器列表
     */
    ThriftTransportPool(int poolSize, int minSize, ServerInfo &server) : ThriftTransportPool(poolSize, minSize, 300, 60, server)
    {
    }

    ThriftTransportPool(int poolSize, ServerInfo &server) : ThriftTransportPool(poolSize, 1, 300, 60, server)
    {
    }

    ThriftTransportPool(ServerInfo &server) : ThriftTransportPool(1, 1, 300, 60, server)
    {
    }

    int getCheckInvervalSecond() override
    {
        return checkInvervalSecond;
    }

    void setCheckInvervalSecond(int checkInvervalSecond) override
    {
        this->checkInvervalSecond = checkInvervalSecond;
    }

    /**
     * 从池中取一个可用连接
     * @return
     */
    std::shared_ptr<ITransportWrapper> get() override
    {
        if (sem_trywait(&access) == 0)
        {
            std::lock_guard<std::mutex> lock(mt);
            for (int i = 0; i < pool.size(); i++)
            {
                if (pool[i]->isAvailable())
                {
                    pool[i]->setIsBusy(true);
                    pool[i]->setLastUseTime(time(0));
                    return pool[i];
                }
            }
            //尝试激活更多连接
            for (int i = 0; i < pool.size(); i++)
            {
                if (!pool[i]->isBusy() && !pool[i]->isDead() && !pool[i]->isOpen())
                {
                    pool[i]->open();
                    pool[i]->setIsBusy(true);
                    pool[i]->setLastUseTime(time(0));
                    return pool[i];
                }
            }
        }
        throw std::runtime_error("can not get available client");
    }

    /**
     * 客户端调用完成后，必须手动调用此方法，将TTransport恢复为可用状态
     *
     * @param client
     */
    void release(ITransportWrapperPtr client) override
    {
        bool released = false;
        {
            std::lock_guard<std::mutex> lock(mt);
            for (int i = 0; i < pool.size(); i++)
            {
                if (client == pool[i] && pool[i]->isBusy())
                {
                    pool[i]->setIsBusy(false);
                    released = true;
                    break;
                }
            }
        }
        if (released)
        {
            sem_post(&access);
        }
    }

    void destory() override
    {
        for (int i = 0; i < pool.size(); i++)
        {
            pool[i]->destroy();
        }
        allowCheck = false;
        checkThread.detach();
        std::cout << "连接池被销毁！" << std::endl;
    }

    /**
     * 获取当前已经激活的连接数
     *
     * @return
     */
    int getActiveCount() override
    {
        int result = 0;
        for (int i = 0; i < pool.size(); i++)
        {
            if (!pool[i]->isDead() && pool[i]->isOpen())
            {
                result += 1;
            }
        }
        return result;
    }

    /**
     * 获取当前繁忙状态的连接数
     *
     * @return
     */
    int getBusyCount() override
    {
        int result = 0;
        for (int i = 0; i < pool.size(); i++)
        {
            if (!pool[i]->isDead() && pool[i]->isBusy())
            {
                result += 1;
            }
        }
        return result;
    }

    /**
     * 获取当前已"挂"掉连接数
     *
     * @return
     */
    int getDeadCount() override
    {
        int result = 0;
        for (int i = 0; i < pool.size(); i++)
        {
            if (pool[i]->isDead())
            {
                result += 1;
            }
        }
        return result;
    }

    string tostring() override
    {
        return "poolsize:" + to_string(pool.size()) +
               ",minSize:" + to_string(minSize) +
               ",maxIdleSecond:" + to_string(maxIdleSecond) +
               ",checkInvervalSecond:" + to_string(checkInvervalSecond) +
               ",active:" + to_string(getActiveCount()) +
               ",busy:" + to_string(getBusyCount()) +
               ",dead:" + to_string(getDeadCount());
    }

    string getWrapperInfo(ITransportWrapperPtr client) override
    {
        for (int i = 0; i < pool.size(); i++)
        {
            if (pool[i] == client)
            {
                return pool[i]->tostring();
            }
        }
        return "";
    }

    /**
     * 连接池初始化
     */
    void init()
    {
        if (sem_init(&access, 0, poolSize))
        {
            printf("Semaphore initialization failed!!\n");
            exit(-1);
        }
        pool.resize(poolSize);

        for (int i = 0; i < pool.size(); i++)
        {
            int j = i % serverInfos.size();
            std::shared_ptr<ITransportPool> transportPool = this->shared_from_this();
            if (i < minSize)
            {
                // 使用可自动回收到连接池的transport
                std::shared_ptr<ITransportWrapper> transportWrapper = std::make_shared<TransportWrapper>(serverInfos[j].getHost(), serverInfos[j].getPort(), true);
                pool[i] = std::make_shared<PoolingTransport>(transportWrapper, transportPool);
            }
            else
            {
                // 使用可自动回收到连接池的transport
                std::shared_ptr<ITransportWrapper> transportWrapper = std::make_shared<TransportWrapper>(serverInfos[j].getHost(), serverInfos[j].getPort());
                pool[i] = std::make_shared<PoolingTransport>(transportWrapper, transportPool);
            }
        }

        check();
    }

private:
    void construct(int poolSize, int minSize, int maxIdleSecond, int checkInvervalSecond, vector<ServerInfo> &serverList)
    {
        if (poolSize <= 0)
        {
            poolSize = 1;
        }
        if (minSize > poolSize)
        {
            minSize = poolSize;
        }
        if (minSize <= 0)
        {
            minSize = 0;
        }
        this->maxIdleSecond = maxIdleSecond;
        this->minSize = minSize;
        this->poolSize = poolSize;
        this->serverInfos = serverList;
        this->allowCheck = true;
        this->checkInvervalSecond = checkInvervalSecond;
    }
    /**
     * 检查空闲连接
     */
    void check()
    {
        checkThread = std::thread([&]()
                                  {
                                      while (allowCheck)
                                      {
                                          std::cout << "--------------" << std::endl;
                                          std::cout << "开始检测空闲连接..." << std::endl;
                                          for (int i = 0; i < pool.size(); i++)
                                          {
                                              if (pool[i] == nullptr)
                                              {
                                                  std::cout << "pool[" << i << "]为null" << std::endl;
                                              }
                                            //   std::cout << pool[i]->tostring() << std::endl;
                                              if (pool[i]->isAvailable())
                                              {
                                                  long idleTime = time(0) - pool[i]->getLastUseTime();
                                                  std::cout << "available pool[" << i << "] idleTime: " << idleTime << std::endl;
                                                  //超过空闲阀值的连接，主动断开，以减少资源消耗
                                                  if (idleTime > maxIdleSecond)
                                                  {
                                                      if (getActiveCount() > minSize)
                                                      {
                                                          pool[i]->destroy();
                                                          pool[i]->setIsBusy(false);
                                                          std::cout << pool[i]->getHost() << ":" << to_string(pool[i]->getPort()) << " 超过空闲时间阀值被断开！" << std::endl;
                                                      }
                                                  }
                                              }
                                          }
                                          std::cout << "当前活动连接数：" << getActiveCount() << std::endl;
                                          std::this_thread::sleep_for(std::chrono::milliseconds(checkInvervalSecond * 1000));
                                      }
                                  });
    }
};

简直太神奇了，以后再也不用担心连接爆了，而且对于不怎么活跃的连接，经过一段时间后，会自动释放掉，连接池也不会经常保持那么多连接，毕竟不用也占有资源。这个框架弄好以后，用起来就非常方便了，这就是一个打地基的过程，后面建起来高楼大厦就更稳更快了，感觉还是要懂设计模式，才能写出大工程的代码，否则新需求来了，维护起来非常痛苦。