模拟一个简单Redis连接池

怼烎@ 2022-03-15 15:38 301阅读 0赞

连接池：简单理解就是一个容器，里面放了一堆的连接对象，这些对象可以用于访问特定的资源，例如JDBC数据库连接池，它有很多连接池的实现，例如：Druid，HikariCP，C3p0，DBCP等等，下面以NoSQL Redis为例，实现一个简单的连接池，供学习参考，如有问题欢迎指正，感谢

首先是连接池接口：

package cn.cuit.study;
    
    import redis.clients.jedis.Jedis;
    
    public interface MyRedisPool {
    	/**
    	 * 连接池初始化
    	 * 
    	 * @param maxTotal
    	 *            最大连接数
    	 * @param maxWaitMillis
    	 *            最长等待超时时间
    	 */
    	public void init(int maxTotal, long maxWaitMillis) throws Exception;
    
    	/**
    	 * 获取连接
    	 * 
    	 * @return
    	 * @throws Exception
    	 */
    	public Jedis getResource() throws Exception;
    
    	/**
    	 * 释放连接
    	 * 
    	 * @return
    	 * @throws Exception
    	 */
    	public Jedis release(Jedis jedis);
    }

连接池接口的实现：

package cn.cuit.study;
    
    import java.util.Objects;
    import java.util.Queue;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    import org.springframework.stereotype.Component;
    
    import redis.clients.jedis.Jedis;
    
    @Component
    public class MyRedisPoolImpl implements MyRedisPool {
    	/**
    	 * 连接池原理
    	 */
    	private int maxTotal;
    	private long maxWaitMillis;
    
    	// 定义一个空闲的连接集合
    	private Queue<Jedis> idleObjects = null;
    	// 定义一个活动的连接集合
    	private Queue<Jedis> activeObjects = null;
    
    	// 总的连接数量，int型 ++操作符非线程安全
    	// private int count;
    	private AtomicInteger count = new AtomicInteger();
    
    	@Override
    	public void init(int maxTotal, long maxWaitMillis) throws Exception {
    		this.maxTotal = maxTotal;
    		this.maxWaitMillis = maxWaitMillis;
    		idleObjects = new LinkedBlockingQueue<Jedis>(maxTotal);
    		activeObjects = new LinkedBlockingQueue<Jedis>(maxTotal);
    	}
    
    	@SuppressWarnings("resource")
    	@Override
    	public Jedis getResource() throws Exception {
    		// 伪代码思路
    		// 0. 记录开始时间戳，用于判断超时时间
    		long startTime = System.currentTimeMillis();
    		// 1.从空闲连接集合中获取连接，如果获取到，将这个连接，放入活动集合中，返回这个连接供获取者使用
    		Jedis jedis = null;
    		while (Objects.isNull(jedis)) {
    			// poll:检索并删除此队列的头部，如果此队列为空，则返回null。
    			jedis = idleObjects.poll();
    			if (Objects.nonNull(jedis)) {
    				activeObjects.offer(jedis); // 元素为null时，offer方法没有异常，add方法会有异常
    				return jedis;
    			}
    			// 2.如果没有从空闲集合中获取到，那么判断连接池是否已满，如果没满，创建一个新的连接，并且放入活动连接，并返回这个连接
    			if (count.get() < maxTotal) {
    				// count++; // ++操作在并发场景是线程不安全的，所以要使用原子类
    				int incrementAndGet = count.incrementAndGet();
    				if (incrementAndGet <= maxTotal) {
    					jedis = new Jedis("127.0.0.1", 6379);
    					System.out.println(">>>>>>>Redis 连接池创建了一个连接" + incrementAndGet);
    					activeObjects.offer(jedis);
    					return jedis;
    				} else {
    					count.decrementAndGet(); // 如果数量超过了最大数量要减下来
    				}
    			}
    			// 3.如果满了，等待其他线程释放连接到空闲连接集合中，如果一定时间内可以得到空闲连接，将连接放入活动集合中，然后返回
    			jedis = ((LinkedBlockingQueue<Jedis>) idleObjects)
    					.poll(maxWaitMillis - (System.currentTimeMillis() - startTime), TimeUnit.MILLISECONDS); // 父类Queue接口poll[移除队列中的元素]方法没有参数，这里可以强转，也可以不用父类，直接用子类
    			// 获取的时候可能会有线程中断异常，所以这里循环处理 【线程中断有优雅处理，另行学习】
    			if (Objects.nonNull(jedis)) {
    				activeObjects.offer(jedis);
    				return jedis;
    			}
    			// 4.如果等待超时，抛出等待超时的异常
    			if (maxWaitMillis < System.currentTimeMillis() - startTime) {
    				throw new Exception("超过了最长等待时间");
    			} else {
    				// 中断了
    				continue;
    			}
    		}
    		return jedis;
    	}
    
    	@Override
    	public Jedis release(Jedis jedis) {
    		if (activeObjects.remove(jedis)) {
    			idleObjects.remove(jedis);
    		}
    		return null;
    	}
    }

测试连接：

package cn.cuit.study;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.CyclicBarrier;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    
    import redis.clients.jedis.Jedis;
    
    @RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration(locations = { "classpath:context/*.xml" })
    public class MyRedisPoolTest {
    
    	@Autowired
    	private MyRedisPool redisPool;
    
    	/**
    	 * 模拟50个并发获取
    	 */
    	private final static int THREAD_NUM = 50;
    
    	/**
    	 * 利用它可以实现类似计数器的功能。比如有一个任务A，它要等待其他4个任务执行完毕之后才能执行，此时就可以利用CountDownLatch来实现这种功能
    	 */
    	private CountDownLatch countDownLatch = new CountDownLatch(THREAD_NUM); // 也类似栅栏，等待指定数量的线程后释放这些线程的等待状态，并发执行
    
    	private CyclicBarrier cyclicBarrier = new CyclicBarrier(THREAD_NUM);
    
    	/**
    	 * 使用CountDownLatch测试接口
    	 * 
    	 * @throws Exception
    	 */
    	@Test
    	public void testRedisPoolInterface() throws Exception {
    		// 连接池的初始化
    		redisPool.init(50, 3000);
    
    		// 模拟一个高并发场景[必须通过多线程，否则就是串行的]，对Redis进行一些操作，目的是分析连接池中是不是只创建了指定的连接数
    		// Jedis jedis = null;
    		for (int i = 0; i < 50; i++) {
    			// jedis = redisPool.getResource();
    			// jedis.incr("pooltest.incr");
    			// redisPool.release(jedis);
    
    			new Thread(new Runnable() {
    				Jedis jedis = null;
    
    				@Override
    				public void run() {
    					try {
    						countDownLatch.await();// 每个线程创建好后都在这里等待，等待一个指令，这个指令可以让所有的线程同时执行
    					} catch (InterruptedException e1) {
    						e1.printStackTrace();
    					}
    					try {
    						jedis = redisPool.getResource();
    						jedis.incr("pooltest.incr");
    
    					} catch (Exception e) {
    						e.printStackTrace();
    					} finally {
    						redisPool.release(jedis);
    					}
    				}
    			}).start();
    			countDownLatch.countDown();// 计数，-1，上面的逻辑就是创建50个线程同时进入等待状态，每创建一个线程这里的计数器就减去1，等减为0【必须】所以循环的次数必须大于countDownLatch的初始值】之后，唤醒所有线程，50个线程就并发执行了
    		}
    
    		Thread.sleep(2000); // 模拟线程延迟
    	}
    
    	/**
    	 * 使用CyclicBarrier测试接口
    	 * 
    	 * @throws Exception
    	 */
    	@Test
    	public void testRedisPoolInterface1() throws Exception {
    		// 连接池的初始化
    		redisPool.init(50, 3000);
    
    		// 模拟一个高并发场景[必须通过多线程，否则就是串行的]，对Redis进行一些操作，目的是分析连接池中是不是只创建了指定的连接数
    		// Jedis jedis = null;
    		for (int i = 0; i < 50; i++) {
    			// jedis = redisPool.getResource();
    			// jedis.incr("pooltest.incr");
    			// redisPool.release(jedis);
    
    			new Thread(new Runnable() {
    				Jedis jedis = null;
    
    				@Override
    				public void run() {
    					try {
    						cyclicBarrier.await();
    					} catch (InterruptedException | BrokenBarrierException e) {
    						e.printStackTrace();
    					} // 每个线程创建好后都在这里等待，等待一个指令，这个指令可以让所有的线程同时执行
    					try {
    						jedis = redisPool.getResource();
    						jedis.incr("pooltest.incr");
    
    					} catch (Exception e) {
    						e.printStackTrace();
    					} finally {
    						redisPool.release(jedis);
    					}
    				}
    			}).start();
    			// 计数，-1，上面的逻辑就是创建50个线程同时进入等待状态，每创建一个线程这里的计数器就减去1，等减为0【必须】所以循环的次数必须大于countDownLatch的初始值】之后，唤醒所有线程，50个线程就并发执行了
    		}
    		Thread.sleep(2000); // 模拟线程延迟
    
    		// 重置栅栏【CyclicBarrier可以重复使用，而CountDownLatch不能】
    		cyclicBarrier.reset();
    	}
    }

测试效果：

#注意，每次测试结果顺序可能会有不同
    >>>>>>>Redis 连接池创建了一个连接3
    >>>>>>>Redis 连接池创建了一个连接50
    >>>>>>>Redis 连接池创建了一个连接1
    >>>>>>>Redis 连接池创建了一个连接46
    >>>>>>>Redis 连接池创建了一个连接43
    >>>>>>>Redis 连接池创建了一个连接48
    >>>>>>>Redis 连接池创建了一个连接42
    >>>>>>>Redis 连接池创建了一个连接44
    >>>>>>>Redis 连接池创建了一个连接45
    >>>>>>>Redis 连接池创建了一个连接47
    >>>>>>>Redis 连接池创建了一个连接49
    >>>>>>>Redis 连接池创建了一个连接41
    >>>>>>>Redis 连接池创建了一个连接39
    >>>>>>>Redis 连接池创建了一个连接38
    >>>>>>>Redis 连接池创建了一个连接40
    >>>>>>>Redis 连接池创建了一个连接37
    >>>>>>>Redis 连接池创建了一个连接34
    >>>>>>>Redis 连接池创建了一个连接32
    >>>>>>>Redis 连接池创建了一个连接30
    >>>>>>>Redis 连接池创建了一个连接29
    >>>>>>>Redis 连接池创建了一个连接35
    >>>>>>>Redis 连接池创建了一个连接36
    >>>>>>>Redis 连接池创建了一个连接31
    >>>>>>>Redis 连接池创建了一个连接33
    >>>>>>>Redis 连接池创建了一个连接28
    >>>>>>>Redis 连接池创建了一个连接27
    >>>>>>>Redis 连接池创建了一个连接26
    >>>>>>>Redis 连接池创建了一个连接25
    >>>>>>>Redis 连接池创建了一个连接24
    >>>>>>>Redis 连接池创建了一个连接22
    >>>>>>>Redis 连接池创建了一个连接23
    >>>>>>>Redis 连接池创建了一个连接21
    >>>>>>>Redis 连接池创建了一个连接20
    >>>>>>>Redis 连接池创建了一个连接19
    >>>>>>>Redis 连接池创建了一个连接16
    >>>>>>>Redis 连接池创建了一个连接17
    >>>>>>>Redis 连接池创建了一个连接18
    >>>>>>>Redis 连接池创建了一个连接15
    >>>>>>>Redis 连接池创建了一个连接14
    >>>>>>>Redis 连接池创建了一个连接13
    >>>>>>>Redis 连接池创建了一个连接8
    >>>>>>>Redis 连接池创建了一个连接10
    >>>>>>>Redis 连接池创建了一个连接12
    >>>>>>>Redis 连接池创建了一个连接11
    >>>>>>>Redis 连接池创建了一个连接9
    >>>>>>>Redis 连接池创建了一个连接7
    >>>>>>>Redis 连接池创建了一个连接5
    >>>>>>>Redis 连接池创建了一个连接2
    >>>>>>>Redis 连接池创建了一个连接6
    >>>>>>>Redis 连接池创建了一个连接4

\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* 不积跬步无以至千里，不积小流无以成江海 \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*