一个简单连接池的实现

一个简单的连接池实现

前言

本人近日正在学习多线程相关的知识，学习线程池的过程中联想到连接池也是一个存在线程安全的复用结构，因此简单探究了连接池相关的实现，该文章用于记录相关的实现思路。

为什么需要连接池？

系统与数据库、缓存进行通信的时候都需要先建立连接、执行相关命令以后断开连接，这个过程将会消耗一定的资源。连接建立需要进行三报文握手，连接断开需要四报文挥手，频繁的建立连接和释放连接将会造成大量的系统资源（主要指CPU资源）浪费。于是连接池便出现了，目的是为了进行连接资源的复用，避免因连接的建立和释放造成系统资源的消耗。此外连接池还能够对连接起到统一的管理，以免在不同的线程中对连接的违规操作（例如连接未正确释放导致端口的占用）。

连接池具有哪些功能？

申请连接

     一个连接池最基本的功能，便是为应用程序提供与其他服务交互的句柄。连接可以通过连接池的初始化提前进行创建，也可以在需要的时候进行创建。创建的连接数量受限于给连接池指定的最大连接数，一旦当前所有的连接都处于使用状态则会进行等待，直到其他业务使用完连接以后归还连接资源。

归还连接

     当一个业务线程使用完资源以后需要将连接归还到线程池中以便于后续的业务操作获取资源。如果线程长时间持有资源或者未归还资源则会造成等待队列的增长，导致系统失去响应。因此正确使用连接池进行资源的归还是十分重要的。

释放连接池

     当系统需要进行停机的时候要能够完成连接资源的统一释放。

     以上三点便是一个连接池最基础的三个功能。本文将会对以上三个功能进行实现，并保证线程的安全性。

连接池的结构

     连接池中应该存在两个逻辑队列和一个连接资源的创建工厂。当连接池中不存在连接时（一般出现于初始化状态）需要调用连接工厂创建生成空闲连接放入到IdleList（空闲列表）中。如果资源被申请使用后应该将连接从IdleList中移除并添加到BusyList，如果使用队列也可以直接使用状态标记的形式来表明资源的占用，本文为了理解方便，在实现过程中都以双列表形式进行处理。当连接被线程使用完以后应该正确归还连接池，需要从BusyList中查找到连接的位置，然后从BusyList中移除加入到IdleList中。

 

 　　　　　　　　　　　　　　　　　　图 1 连接池的申请和归还过程

连接池的操作

本小节将着重于连接池的具体操作进行讲解。描述连接池的初始化，服务过程和释放过程。

连接池初始化

     连接池在初始化的时候应该提供初始化时的连接建立个数以及最大连接数量用于进行连接的初始化创建和最大连接数量的创建。初始的连接个数不应该超过最大的连接个数。当连接池初始化完成后才能正式向其他线程提供服务。

连接申请

     线程申请连接时应该先检查连接池中是否存在空闲的连接，如果存在空闲连接则使用空闲连接并加入到BusyList当中。如果因为空闲连接不足则要检查连接池当前管理的连接数量是否以及到达最大连接数量，如果没有则直接创建新的连接加入到BusyList。如果以上条件都不满足的情况应该进入到等待状态，等待连接归还的通知然后重新进行资源的竞争。

 

 　　　　　　　　　　　　　　　　　　图 2 连接申请过程

连接归还

     连接归还时应该通知因无法获取资源导致等待的线程，让等待的线程重新发起对资源的竞争。为了避免饥饿，这个过程可以使用公平锁来进行处理。

连接池释放

     当系统将要关闭的时候应该等待其他线程对资源的归还，保证其他线程能够正常完成业务的执行。当归还完成后集中进行资源的释放。

接口定义

     根据上文的功能，需要简单的定义如下方法：

Init	连接池初始化
GetResource	获取连接
ReturnResource	归还连接
Close	关闭连接池

具体实现

接口

连接定义：

package pool;

/**
 * <p>
 * description: 可被连接池管理的连接
 * </p>
 *
 * @author fruitdish
 * @date 5/30/2020
 * @since v1.0.11
 */
public interface Connection {
  boolean open();

  boolean close();
}

连接工厂定义：

package pool;

/**
 * <p>
 * description: 用于创建连接
 * </p>
 *
 * @author fruitdish
 * @date 5/30/2020
 * @since v1.0.11
 */
public interface ConnectionFactory {
  Connection create() throws Exception;
}

　　

连接池定义：

package pool;

/**
 * <p>
 * description:
 * 连接池的接口定义
 * </p>
 *
 * @author fruitdish
 * @date 5/29/2020
 * @since v1.0.11
 */
public interface ConnectionPool {

  /**
   * 定义连接的初始化数量和最大活跃连接，将初始化的连接全部放入到IdleList中
   *
   * @param initNum   初始化数量
   * @param maxActive 最大活跃连接数
   */
  void init(int initNum, int maxActive) throws Exception;

  /**
   * 从连接池中获取连接资源。如果因为无法获取资源则进入到等待状态
   * @return
   */
  Connection getResource() throws Exception;

  /**
   * 当连接使用完以后对连接进行归还
   * @param resource 待归还连接
   */
  void release(Connection resource);

  /**
   * 关闭所有的连接
   * 如果有连接正在使用，则进行等待
   */
  void close() throws InterruptedException;
}

实现：

package pool;
import java.util.LinkedList;
import java.util.List;

/**
 * <p>
 * description:
 * </p>
 *
 * @author fruitdish
 * @date 5/29/2020
 * @since v1.0.11
 */
public class JedisPool implements ConnectionPool {
  /**
   * busylist，所有正在使用的连接都会存放在该链表中
   */
  private List<Connection> busyList = new LinkedList<>();
  /**
   * idleList,所有空闲的连接都会存放在该链表中
   */
  private List<Connection> idleList = new LinkedList<>();
  /**
   * 保证线程安全的锁对象
   */
  private Object lock =new Object();
  /**
   * 最大连接数量
   * 如果当前连接数量已经等于最大连接数量，则需要等待
   */
  private int maxActive = 0;
  /**
   * 连接计数
   */
  private int active = 0;
  /**
   * 连接池的运行状态
   */
  private int status = NO_INIT;
  /**
   * 未初始化状态，处于该状态的连接池不能提供正常的服务
   */
  private static final int NO_INIT= -1;
  /**
   * 对连接池的初始化已经成功，可以正常提供服务
   */
  private static final int SERVICE = 0;
  /**
   * 连接池处于关闭状态，不能提供连接获取功能，但是连接归还能够正常执行
   */
  private static final int CLOSED=1;

  private ConnectionFactory connectionFactory = new ConnectionFactory() {
    @Override
    public Connection create() throws Exception {
      return new Connection() {
        @Override
        public boolean open() {
          return false;
        }

        @Override
        public boolean close() {
          return false;
        }
      };
    }
  };

  public JedisPool() {
  }

  public JedisPool(ConnectionFactory connectionFactory) {
    this.connectionFactory = connectionFactory;
  }

  @Override
  public void init(int initNum, int maxActive) throws Exception {
    if (initNum>maxActive) {
      throw new Exception("初始化数量大于最大连接数量");
    }
    synchronized (lock) {
      this.maxActive = maxActive;
      if (status == SERVICE) {
        return;
      }
      for (int i=0;i<initNum;i++) {
        Connection c = connectionFactory.create();
        idleList.add(c);
        active++;
      }
      // 转为服务状态
      status = SERVICE;
    }
  }

  @Override
  public Connection getResource() throws Exception {
    for (;;) {
     synchronized (lock) {
       if (status != SERVICE) {
         throw new Exception("连程池无法提供服务");
       }
       if (idleList.isEmpty()) {
         if (active<maxActive) {
           Connection c = connectionFactory.create();
           busyList.add(c);
           active++;
           return c;
         } else {
           lock.wait();
           // 被唤醒后一定会有连接可用
           Connection c = idleList.remove(0);
           busyList.add(c);
           return c;
         }
       }else {
         Connection c = idleList.remove(0);
         busyList.add(c);
         return c;
       }
     }
    }
  }

  @Override
  public void release(Connection resource) {
    synchronized (lock) {
      busyList.remove(resource);
      idleList.add(resource);
      lock.notify();
    }
  }

  @Override
  public void close() throws InterruptedException {
    synchronized (lock) {
      status = CLOSED;
      while (!busyList.isEmpty()) {
        System.out.println("连接未归还，等待");
        lock.wait();
      }
      for (Connection c : idleList) {
        c.close();
      }
      System.out.println("连接池关闭");
    }
  }
}