likes
comments
collection
share

线程间相互协作：等待超时模式实现一个连接池

作者站长头像
站长
· 阅读数 29

线程间相互协作：等待超时模式实现一个连接池

  • 场景：在调用SM程序的时候需要配置连接池，但是这个连接池里面有一个超时连接

    • 如果说在获取连接的时候，超过了时间限制，那么就会抛出一个异常信息出来

    • 在等待和通知模式基础上，搞一个这个；这个在Spring，MyBatist里面就是用的这个

      • 调用一个方法后等待一段时间，一般是固定时间，例如在连接池里面配置成5s
      • 调用在等待时间内完成调用，就正常返回，如果说超时了，就抛出异常或者返回一个默认值
      • 在C3P0,DBCP 等连接池中都有

  • 处理：手动实现连接池(基于JDBC)

    • 基础

      • 实现了Collection接口，JDBC的东西(实现部分功能)
      • 数据库一般情况下是有最大连接数的，跟OS创建线程一样

    • 实现：

      1. 连接池(DBPool)存放连接：就是一个容器

    • 细节：

      • 为什么要用poo调用wait

        • 因为是在这个容器上等待，当池子为空或者为满的时候去通知其他的线程

    • 运行结果：缺失是实现了等待超时模式，很不错

      线程间相互协作：等待超时模式实现一个连接池

    • DBPool.java：数据库连接池

       package cn.enjoyedu.ch1.pool;
 ​
 import java.sql.Connection;
 import java.util.LinkedList;
 ​
 /**
  *类说明：连接池的实现
  */
 public class DBPool {
 ​
     private static LinkedList<Connection> pool = new LinkedList<Connection>();
 ​
     //限制了线程的最大连接数，一般还有最小连接，最大连接数定义为20
     public DBPool(int initialSize) {
         if (initialSize > 0) {
             for (int i = 0; i < initialSize; i++) {
                 pool.addLast(SqlConnectImpl.fetchConnection());
             }
         }
     }
 //释放连接，用完后放回连接池，通知其他等待连接的应用
     public void releaseConnection(Connection connection) {
         if (connection != null) {
             //TODO这个池子不空了，通知其他等待连接的线程
             synchronized (pool){
                 pool.addLast(connection);
                 //通知其他等待连接的线程
                 pool.notifyAll();
             }
         }
     }
 ​
     // 在mills(超时时长)内无法获取到连接，将会返回null，真正的还要抛出异常去
     //数据库连接池的大体实现思路
     //获取连接
     public Connection fetchConnection(long mills) throws InterruptedException {
         //TODO同一时刻只能有一个线程操作数据库
         synchronized (pool){
             //永不超时
             if(mills<0){
                 //去拿连接
                 while(pool.isEmpty()){
                     wait();
                 }
                 //从尾巴放，从头拿
                 return pool.removeFirst();
             }else{
                 //获取超时的时刻，什么时候超时
                 long furture = System.currentTimeMillis()+mills;
                 //需要等待多久
                 long remaining = mills;
                 while(pool.isEmpty()&&remaining>0){
                     //这个时候需要用pool调用wait方法：注意wait的调用
                     pool.wait(remaining);//等那么久
                     //每被唤醒后，需要重新计算等待时间(因为有可能是抢不到的)
                     remaining = furture-System.currentTimeMillis();
                 }
                 //跳出这个while循环，那么就是两种情况(要么成功拿到CPU--->去连接池里面拿连接，
                 // 要么等待时间结束了--->返回一个空的连接，返回调用者，告诉他，他在没有抢到并且时间结束了)
                 Connection connection = null;
                 //当抢到了CPU，这个时候就要去连接池里面拿东西了同时排除了那种因为超时进来的；
                 if (!pool.isEmpty()){
                     connection = pool.removeFirst();
                 }
                 return connection;
             }
         }
     }
 }

    • DBTest.java

       package cn.enjoyedu.ch1.pool;
 ​
 import java.sql.Connection;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 ​
 /**
  *类说明：
  */
 public class DBPoolTest {
     static DBPool pool  = new DBPool(20);
     // 控制器:控制main线程将会等待所有Woker结束后才能继续执行，控制所有线程一起执行工作
     static CountDownLatch end;
 ​
     //定义了50个线程，每一个线程尝试去数据库连接池中拿20次
     public static void main(String[] args) throws Exception {
         // 线程数量
         int threadCount = 50;
         end = new CountDownLatch(threadCount);
         int count = 20;//每个线程的操作次数
         //原子变量，保证线程安全性
         AtomicInteger got = new AtomicInteger();//计数器：统计可以拿到连接的线程
         AtomicInteger notGot = new AtomicInteger();//计数器：统计没有拿到连接的线程
 ​
         for (int i = 0; i < threadCount; i++) {
             Thread thread = new Thread(new Worker(count, got, notGot), 
                     "worker_"+i);
             thread.start();
         }
 ​
         end.await();// main线程在此处等待
 ​
         System.out.println("总共尝试了: " + (threadCount * count));
         System.out.println("拿到连接的次数：  " + got);
         System.out.println("没能连接的次数： " + notGot);
     }
 ​
     static class Worker implements Runnable {
         int           count;
         AtomicInteger got;
         AtomicInteger notGot;
 ​
         public Worker(int count, AtomicInteger got,
                                AtomicInteger notGot) {
             this.count = count;
             this.got = got;
             this.notGot = notGot;
         }
 ​
         public void run() {
             while (count > 0) {
                 try {
                     // 从线程池中获取连接，如果1000ms内无法获取到，将会返回null
                     // 分别统计连接获取的数量got和未获取到的数量notGot
                     Connection connection = pool.fetchConnection(1000);
                     if (connection != null) {
                         try {
                             //拿到连接后，进行数据库操作
                             connection.createStatement();
                             connection.commit();
                         } finally {
                             //连接耍完了，那么就将这个连接重新放回数据库连接池
                             pool.releaseConnection(connection);
                             //统计这种成功的次数
                             got.incrementAndGet();
                         }
                     } else {
                         //当这个拿到的连接时空的，就统计这些失败后的问题
                         notGot.incrementAndGet();
                         System.out.println(Thread.currentThread().getName()
                                 +"等待超时!");
                     }
                 } catch (Exception ex) {
                 } finally {
                     count--;
                 }
             }
             end.countDown();
         }
     }
 }

    • SqlConnectImpl.java 重写了大量的方法

       ```
  package cn.enjoyedu.ch1.pool;

     import cn.enjoyedu.tools.SleepTools;

     import java.sql.*;
     import java.util.Map;
     import java.util.Properties;
     import java.util.concurrent.Executor;

     /**
      *类说明：
      */
     public class SqlConnectImpl implements Connection{

     /*拿一个数据库连接*/
 public static final Connection fetchConnection(){
     return new SqlConnectImpl();
 }

 @Override
 public boolean isWrapperFor(Class<?> arg0) throws SQLException {
         // TODO Auto-generated method stub
         return false;
 }

 @Override
 public <T> T unwrap(Class<T> arg0) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public void abort(Executor arg0) throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void clearWarnings() throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void close() throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void commit() throws SQLException {
         SleepTools.ms(70);
 }

 @Override
 public Array createArrayOf(String arg0, Object[] arg1) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public Blob createBlob() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public Clob createClob() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public NClob createNClob() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public SQLXML createSQLXML() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public Statement createStatement() throws SQLException {
         SleepTools.ms(1);
         return null;
 }

 @Override
 public Statement createStatement(int arg0, int arg1) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public Statement createStatement(int arg0, int arg1, int arg2) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public Struct createStruct(String arg0, Object[] arg1) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public boolean getAutoCommit() throws SQLException {
         // TODO Auto-generated method stub
         return false;
 }

 @Override
 public String getCatalog() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public Properties getClientInfo() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public String getClientInfo(String arg0) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public int getHoldability() throws SQLException {
         // TODO Auto-generated method stub
         return 0;
 }

 @Override
 public DatabaseMetaData getMetaData() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public int getNetworkTimeout() throws SQLException {
         // TODO Auto-generated method stub
         return 0;
 }

 @Override
 public String getSchema() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public int getTransactionIsolation() throws SQLException {
         // TODO Auto-generated method stub
         return 0;
 }

 @Override
 public Map<String, Class<?>> getTypeMap() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public SQLWarning getWarnings() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public boolean isClosed() throws SQLException {
         // TODO Auto-generated method stub
         return false;
 }

 @Override
 public boolean isReadOnly() throws SQLException {
         // TODO Auto-generated method stub
         return false;
 }

 @Override
 public boolean isValid(int arg0) throws SQLException {
         // TODO Auto-generated method stub
         return false;
 }

 @Override
 public String nativeSQL(String arg0) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public CallableStatement prepareCall(String arg0) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public CallableStatement prepareCall(String arg0, int arg1, int arg2) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public CallableStatement prepareCall(String arg0, int arg1, int arg2, int arg3) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public PreparedStatement prepareStatement(String arg0) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public PreparedStatement prepareStatement(String arg0, int arg1) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public PreparedStatement prepareStatement(String arg0, int[] arg1) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public PreparedStatement prepareStatement(String arg0, String[] arg1) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public PreparedStatement prepareStatement(String arg0, int arg1, int arg2) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public PreparedStatement prepareStatement(String arg0, int arg1, int arg2, int arg3) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public void releaseSavepoint(Savepoint arg0) throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void rollback() throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void rollback(Savepoint arg0) throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void setAutoCommit(boolean arg0) throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void setCatalog(String arg0) throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void setClientInfo(Properties arg0) throws SQLClientInfoException {
         // TODO Auto-generated method stub

 }

 @Override
 public void setClientInfo(String arg0, String arg1) throws SQLClientInfoException {
         // TODO Auto-generated method stub

 }

 @Override
 public void setHoldability(int arg0) throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void setNetworkTimeout(Executor arg0, int arg1) throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void setReadOnly(boolean arg0) throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public Savepoint setSavepoint() throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public Savepoint setSavepoint(String arg0) throws SQLException {
         // TODO Auto-generated method stub
         return null;
 }

 @Override
 public void setSchema(String arg0) throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void setTransactionIsolation(int arg0) throws SQLException {
         // TODO Auto-generated method stub

 }

 @Override
 public void setTypeMap(Map<String, Class<?>> arg0) throws SQLException {
         // TODO Auto-generated method stub

 }

 }
 ```

  • 方法对锁的影响

    • yield()

      • 概述：调用这个只是让出了CPU的执行权，不会释放当前线程持有的锁

    • sleep()

      • 概述：线程休眠，不会释放锁，

      • 场景：

        • 一个线程，休眠，先拿到锁
        • 另外一个线程，不休眠去拿锁

      • 结果：

        • 不休眠的线程只有等休眠线程休眠完成并执行完同步代码块后才能拿到锁

      • 运行截图：

        线程间相互协作：等待超时模式实现一个连接池

      • 测试代码：

         package cn.enjoyedu.ch1.base;
 ​
 /**
  *类说明：测试Sleep对锁的影响
  */
 public class SleepLock {
     private Object lock = new Object();
 ​
     public static void main(String[] args) {
         SleepLock sleepTest = new SleepLock();
         Thread threadA = sleepTest.new ThreadSleep();
         threadA.setName("ThreadSleep");
         Thread threadB = sleepTest.new ThreadNotSleep();
         threadB.setName("ThreadNotSleep");
         threadA.start();
         try {
             Thread.sleep(1000);
             System.out.println(" Main slept!");
         } catch (InterruptedException e) {
             e.printStackTrace();
         }
         threadB.start();
     }
 ​
     private class ThreadSleep extends Thread{
 ​
         @Override
         public void run() {
             String threadName = Thread.currentThread().getName();
             System.out.println(threadName+" will take the lock");
             try {
 ​
                 synchronized(lock) {
                     System.out.println(threadName+" taking the lock");
                     Thread.sleep(5000);
                     System.out.println("Finish the work: "+threadName);
                 }
             } catch (InterruptedException e) {
                 //e.printStackTrace();
             }
         }
     }
 ​
     private class ThreadNotSleep extends Thread{
 ​
         @Override
         public void run() {
             String threadName = Thread.currentThread().getName();
             System.out.println(threadName+" will take the lock time="+System.currentTimeMillis());
             synchronized(lock) {
                 System.out.println(threadName+" taking the lock time="+System.currentTimeMillis());
                 System.out.println("Finish the work: "+threadName);
             }
         }
     }
 }

    • wait()

      • 线程调用wait后会释放自己的锁，当被唤醒后，竞争锁，竞争到才执行后面的代码

    • notify/notifyAll

      • 对于锁没有影响，一般等到同步代码块执行完成后，才释放锁
      • 在使用过程中，一般将其放在最后一行；
转载自:https://juejin.cn/post/7067229482179887111
评论
请登录