自定义带监控的数据库连接池

阳光穿透心脏的1/2处 2022-12-30 14:57 192阅读 0赞

我们可以先看一个最基础的jdbc连接的使用方法，在这里我们使用同一个`connection`做了两次查询的操作

public static void main(String[] args) throws ClassNotFoundException, SQLException { 
        Class.forName("com.mysql.jdbc.Driver");
        Connection connection = DriverManager.getConnection(
          "jdbc:mysql://localhost:3306/database?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true",
          "root",
          "root"
        );
        // sql执行
        PreparedStatement preparedStatement = connection.prepareStatement(
          "select * from table_a;"
        );
        // 获取返回结果
        ResultSet resultSet = preparedStatement.executeQuery();
        while (resultSet.next()) { 
          System.out.println(resultSet.getString("id"));
          System.out.println(resultSet.getString("user_id"));
        }
        resultSet.close();
        preparedStatement.close();
    
    
        PreparedStatement preparedStatement2 = connection.prepareStatement(
          "select * from table_b"
        );
        ResultSet resultSet2 = preparedStatement2.executeQuery();
        while (resultSet2.next()) { 
          System.out.println(resultSet2.getString(1));
          System.out.println(resultSet2.getString(2));
        }
        resultSet2.close();
        preparedStatement2.close();
    
        connection.close();
    
    }

我们如何实现连接池呢，有以下几步：

1.  实现`PooledObjectFactory<Connection>`

public class ConnectionPooledObjectFactory implements PooledObjectFactory<Connection> { 
    
      @Override
      public PooledObject<Connection> makeObject() throws Exception { 
        Class.forName("com.mysql.jdbc.Driver");
        Connection connection = DriverManager.getConnection(
          "jdbc:mysql://localhost:3306/database?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true",
          "root",
          "root"
        );
        return new DefaultPooledObject<>(connection);
      }
    
      @Override
      public void destroyObject(PooledObject<Connection> p) throws Exception { 
        p.getObject().close();
      }
    
      @Override
      public boolean validateObject(PooledObject<Connection> p) { 
        Connection connection = p.getObject();
        try { 
          PreparedStatement statement = connection.prepareStatement("SELECT 1");
          ResultSet resultSet = statement.executeQuery();
          int i = resultSet.getInt(1);
          return i == 1;
        } catch (SQLException e) { 
          return false;
        }
      }
    
      @Override
      public void activateObject(PooledObject<Connection> p) throws Exception { 
        // 可以把connection额外的配置放到这里
      }
    
      @Override
      public void passivateObject(PooledObject<Connection> p) throws Exception { 
        
      }
    }

1.  实现`DataSource`

public class MyDataSource implements DataSource { 
      private GenericObjectPool<MyConnection> pool;
    
      public GenericObjectPool<MyConnection> getPool() { 
        return pool;
      }
    
      public void setPool(GenericObjectPool<MyConnection> pool) { 
        this.pool = pool;
      }
    
      public MyDataSource() { 
        ConnectionPooledObjectFactory factory = new ConnectionPooledObjectFactory();
        this.pool = new GenericObjectPool<>(factory);
      }
    
      @Override
      public Connection getConnection() throws SQLException { 
        try { 
          return this.pool.borrowObject();
        } catch (Exception e) { 
          throw new SQLException("获取连接失败！", e);
        }
      }
    
      @Override
      public Connection getConnection(String username, String password) throws SQLException { 
        return this.getConnection();
      }
    
      @Override
      public <T> T unwrap(Class<T> iface) throws SQLException { 
        throw new UnsupportedOperationException("不支持的操作！");
      }
    
      @Override
      public boolean isWrapperFor(Class<?> iface) throws SQLException { 
        throw new UnsupportedOperationException("不支持的操作！");
      }
    
      @Override
      public PrintWriter getLogWriter() throws SQLException { 
        throw new UnsupportedOperationException("不支持的操作！");
      }
    
      @Override
      public void setLogWriter(PrintWriter out) throws SQLException { 
        throw new UnsupportedOperationException("不支持的操作！");
      }
    
      @Override
      public void setLoginTimeout(int seconds) throws SQLException { 
        throw new UnsupportedOperationException("不支持的操作！");
      }
    
      @Override
      public int getLoginTimeout() throws SQLException { 
        throw new UnsupportedOperationException("不支持的操作！");
      }
    
      @Override
      public Logger getParentLogger() throws SQLFeatureNotSupportedException { 
        throw new UnsupportedOperationException("不支持的操作！");
      }
    }

1.  修改启动类，修改dataSource

public class Application { 
    
        public static void main(String[] args) { 
            SpringApplication.run(Application.class, args);
        }
        
        @Bean
        @Primary
        public MyDataSource dataSource(){ 
            return new MyDataSource();
        }
    }

然后我们启动项目，尝试连接数据库，这里能正常获取到数据就表示已经配置成功了！

然后我们实现`监控`功能，我们新建一个`DataSourceEnpoint`类，使用`@Endpoint(id = "datasource")`注解，意思是我们可以使用`/actuator/datasource`

@Endpoint(id = "datasource")
    public class DataSourceEnpoint { 
      private MyDataSource dataSource;
    
      public DataSourceEnpoint(MyDataSource dataSource) { 
        this.dataSource = dataSource;
      }
    
      @ReadOperation
      public Map<String, Object> pool() { 
        GenericObjectPool<MyConnection> pool = dataSource.getPool();
        HashMap<String, Object> map = Maps.newHashMap();
        map.put("numActive", pool.getNumActive());
        map.put("numIdle", pool.getNumIdle());
        map.put("createdCount", pool.getCreatedCount());
        return map;
      }
    }

修改启动类

public class Application { 
    
        public static void main(String[] args) { 
            SpringApplication.run(Application.class, args);
        }
        
        @Bean
        @Primary
        public DataSource dataSource(){ 
            return new MyDataSource();
        }
    
        @Bean
        public DataSourceEnpoint dataSourceEnpoint() { 
            DataSource dataSource = this.dataSource();
            return new DataSourceEnpoint((MyDataSource) dataSource);
        }
    }

这里会有一个类型转换的报错，解决方案是在yml文件中，让`javamelody`禁用`dataSource`

javamelody:
      excluded-datasources: dataSource

然后重启项目，这个时候我们访问几次数据库，再访问`/actuator/datasource`，能看到我们创建了6个对象，活跃6个对象，以及0个空闲，说明我们的connection没有进行回收。  
![在这里插入图片描述][20201226162826973.png]  
那如何解决connection回收的问题呢，这个时候我们再创建一个`MyConnection`

// 1. 回收statement
    // 2. 回收resultSet
    // 3. 复用Connection
    public class MyConnection implements Connection { 
      private Connection connection;
      private Statement statement;
      private ObjectPool<MyConnection> objectPool;
    
      public ObjectPool<MyConnection> getObjectPool() { 
        return objectPool;
      }
    
      public void setObjectPool(ObjectPool<MyConnection> objectPool) { 
        this.objectPool = objectPool;
      }
    
      public Connection getConnection() { 
        return connection;
      }
    
      public void setConnection(Connection connection) { 
        this.connection = connection;
      }
    
      public Statement getStatement() { 
        return statement;
      }
    
      public void setStatement(Statement statement) { 
        this.statement = statement;
      }
    
      @Override
      public Statement createStatement() throws SQLException { 
        return this.connection.createStatement();
      }
    
      @Override
      public PreparedStatement prepareStatement(String sql) throws SQLException { 
        PreparedStatement prepareStatement = this.connection.prepareStatement(sql);
        this.statement = prepareStatement;
        return prepareStatement;
      }
    
      @Override
      public CallableStatement prepareCall(String sql) throws SQLException { 
        return this.connection.prepareCall(sql);
      }
    
      @Override
      public String nativeSQL(String sql) throws SQLException { 
        return this.connection.nativeSQL(sql);
      }
    
      @Override
      public void setAutoCommit(boolean autoCommit) throws SQLException { 
        this.connection.setAutoCommit(autoCommit);
      }
    
      @Override
      public boolean getAutoCommit() throws SQLException { 
        return this.connection.getAutoCommit();
      }
    
      @Override
      public void commit() throws SQLException { 
        this.connection.commit();
      }
    
      @Override
      public void rollback() throws SQLException { 
        this.connection.rollback();
      }
    
      @Override
      public void close() throws SQLException { 
        // 如果底层的Connection已经关闭
        if (this.isClosed()) { 
          try { 
            objectPool.invalidateObject(this);
          } catch (Exception e) { 
            throw new SQLException(e);
          }
        }
        // 底层Connection没有关闭，可以继续复用
        else { 
          try { 
            objectPool.returnObject(this);
          } catch (Exception e) { 
            this.connection.close();
            throw new SQLException(e);
          }
        }
      }
    
      @Override
      public boolean isClosed() throws SQLException { 
        return this.connection.isClosed();
      }
    
      @Override
      public DatabaseMetaData getMetaData() throws SQLException { 
        return this.connection.getMetaData();
      }
    
      @Override
      public void setReadOnly(boolean readOnly) throws SQLException { 
        this.connection.setReadOnly(readOnly);
      }
    
      @Override
      public boolean isReadOnly() throws SQLException { 
        return this.connection.isReadOnly();
      }
    
      @Override
      public void setCatalog(String catalog) throws SQLException { 
        this.connection.setCatalog(catalog);
      }
    
      @Override
      public String getCatalog() throws SQLException { 
        return this.connection.getCatalog();
      }
    
      @Override
      public void setTransactionIsolation(int level) throws SQLException { 
        this.connection.setTransactionIsolation(level);
      }
    
      @Override
      public int getTransactionIsolation() throws SQLException { 
        return this.connection.getTransactionIsolation();
      }
    
      @Override
      public SQLWarning getWarnings() throws SQLException { 
        return this.connection.getWarnings();
      }
    
      @Override
      public void clearWarnings() throws SQLException { 
        this.connection.clearWarnings();
      }
    
      @Override
      public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException { 
        Statement statement = this.connection.createStatement(resultSetType, resultSetConcurrency);
        this.statement = statement;
        return statement;
      }
    
      @Override
      public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException { 
        PreparedStatement preparedStatement = this.connection.prepareStatement(sql, resultSetType, resultSetConcurrency);
        this.statement = preparedStatement;
        return preparedStatement;
      }
    
      @Override
      public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException { 
        return this.connection.prepareCall(sql, resultSetType, resultSetConcurrency);
      }
    
      @Override
      public Map<String, Class<?>> getTypeMap() throws SQLException { 
        return this.connection.getTypeMap();
      }
    
      @Override
      public void setTypeMap(Map<String, Class<?>> map) throws SQLException { 
        this.connection.setTypeMap(map);
      }
    
      @Override
      public void setHoldability(int holdability) throws SQLException { 
        this.connection.setHoldability(holdability);
      }
    
      @Override
      public int getHoldability() throws SQLException { 
        return this.connection.getHoldability();
      }
    
      @Override
      public Savepoint setSavepoint() throws SQLException { 
        return this.connection.setSavepoint();
      }
    
      @Override
      public Savepoint setSavepoint(String name) throws SQLException { 
        return this.connection.setSavepoint(name);
      }
    
      @Override
      public void rollback(Savepoint savepoint) throws SQLException { 
        this.connection.rollback(savepoint);
      }
    
      @Override
      public void releaseSavepoint(Savepoint savepoint) throws SQLException { 
        this.connection.releaseSavepoint(savepoint);
      }
    
      @Override
      public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException { 
        Statement statement = this.connection.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability);
        this.statement = statement;
        return statement;
      }
    
      @Override
      public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException { 
        PreparedStatement statement = this.connection.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
        this.statement = statement;
        return statement;
      }
    
      @Override
      public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException { 
        return this.connection.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
      }
    
      @Override
      public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException { 
        PreparedStatement statement = this.connection.prepareStatement(sql, autoGeneratedKeys);
        this.statement = statement;
        return statement;
      }
    
      @Override
      public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException { 
        PreparedStatement statement = this.connection.prepareStatement(sql, columnIndexes);
        this.statement = statement;
        return statement;
      }
    
      @Override
      public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException { 
        PreparedStatement statement = this.connection.prepareStatement(sql, columnNames);
        this.statement = statement;
        return statement;
      }
    
      @Override
      public Clob createClob() throws SQLException { 
        return this.connection.createClob();
      }
    
      @Override
      public Blob createBlob() throws SQLException { 
        return this.connection.createBlob();
      }
    
      @Override
      public NClob createNClob() throws SQLException { 
        return this.connection.createNClob();
      }
    
      @Override
      public SQLXML createSQLXML() throws SQLException { 
        return this.connection.createSQLXML();
      }
    
      @Override
      public boolean isValid(int timeout) throws SQLException { 
        return this.connection.isValid(timeout);
      }
    
      @Override
      public void setClientInfo(String name, String value) throws SQLClientInfoException { 
        this.connection.setClientInfo(name, value);
      }
    
      @Override
      public void setClientInfo(Properties properties) throws SQLClientInfoException { 
        this.connection.setClientInfo(properties);
      }
    
      @Override
      public String getClientInfo(String name) throws SQLException { 
        return this.connection.getClientInfo(name);
      }
    
      @Override
      public Properties getClientInfo() throws SQLException { 
        return this.connection.getClientInfo();
      }
    
      @Override
      public Array createArrayOf(String typeName, Object[] elements) throws SQLException { 
        return this.connection.createArrayOf(typeName, elements);
      }
    
      @Override
      public Struct createStruct(String typeName, Object[] attributes) throws SQLException { 
        return this.connection.createStruct(typeName, attributes);
      }
    
      @Override
      public void setSchema(String schema) throws SQLException { 
        this.connection.setSchema(schema);
      }
    
      @Override
      public String getSchema() throws SQLException { 
        return this.connection.getSchema();
      }
    
      @Override
      public void abort(Executor executor) throws SQLException { 
        this.connection.abort(executor);
      }
    
      @Override
      public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException { 
        this.connection.setNetworkTimeout(executor, milliseconds);
      }
    
      @Override
      public int getNetworkTimeout() throws SQLException { 
        return this.connection.getNetworkTimeout();
      }
    
      @Override
      public <T> T unwrap(Class<T> iface) throws SQLException { 
        return this.connection.unwrap(iface);
      }
    
      @Override
      public boolean isWrapperFor(Class<?> iface) throws SQLException { 
        return this.connection.isWrapperFor(iface);
      }
    
    }

然后将`ConnectionPooledObjectFactory`中的`Connecrion`全部修改为`MyConnecrion`,并补充`passivateObject`的逻辑

public class ConnectionPooledObjectFactory implements PooledObjectFactory<MyConnection> { 
      private ObjectPool<MyConnection> objectPool;
    
      public ObjectPool<MyConnection> getObjectPool() { 
        return objectPool;
      }
    
      public void setObjectPool(ObjectPool<MyConnection> objectPool) { 
        this.objectPool = objectPool;
      }
    
      @Override
      public PooledObject<MyConnection> makeObject() throws Exception { 
        Class.forName("com.mysql.jdbc.Driver");
        Connection connection = DriverManager.getConnection(
          "jdbc:mysql://localhost:3306/database?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true",
          "root",
          "root"
        );
        MyConnection myConnection = new MyConnection();
        myConnection.setConnection(connection);
        myConnection.setObjectPool(objectPool);
        return new DefaultPooledObject<>(myConnection);
      }
    
      @Override
      public void destroyObject(PooledObject<MyConnection> p) throws Exception { 
        p.getObject().close();
      }
    
      @Override
      public boolean validateObject(PooledObject<MyConnection> p) { 
        Connection connection = p.getObject();
        try { 
          PreparedStatement statement = connection.prepareStatement("SELECT 1");
          ResultSet resultSet = statement.executeQuery();
          int i = resultSet.getInt(1);
          return i == 1;
        } catch (SQLException e) { 
          return false;
        }
      }
    
      @Override
      public void activateObject(PooledObject<MyConnection> p) throws Exception { 
        // 可以把connection额外的配置放到这里
      }
    
      @Override
      public void passivateObject(PooledObject<MyConnection> p) throws Exception { 
        // 钝化
        MyConnection myConnection = p.getObject();
        Statement statement = myConnection.getStatement();
        if (statement != null) { 
          statement.close();
        }
      }
    }

然后修改一下`MyDataSource`，给factory设一下值

public MyDataSource() { 
        ConnectionPooledObjectFactory factory = new ConnectionPooledObjectFactory();
        this.pool = new GenericObjectPool<>(factory);
        factory.setObjectPool(pool);
      }

启动项目，再次访问`/actuator/datasource`，可以看到这个时候对象已经可以正常复用了，变成了一个空闲的连接  
![在这里插入图片描述][20201226164524379.png]  
到这里，一个带监控的数据库连接池就实现完成了！

[20201226162826973.png]: /images/20221120/59b1ca3adc184c3c805365ca33f57a66.png
[20201226164524379.png]: /images/20221120/3463e44dd65d47f5a1020c1d8e0b4553.png