最近线上数据库迁到haproxy上，突然出现了很多数据库连接失败的错误，经过排查是因为我们使用了Mysql的ReplicationDriver,数据库连接池使用的是druid, 而druid针对Mysql Replication 连接的检查实现上有个bug ,导致的。当我正准备提issue的时候，发现很多人都遇到了这个问题，所以想写遍文章记录一下。话不多说，下面我们来正式复盘一下这个问题和定位解决的过程。
之前公司的DBA新上线了一套HAProxy用来替代之前的VIP的高可用法案，所以我们也将从库的连接从VIP迁到了HAProxy.但是上线后不久我们发现线上开始出现数据库连接错误，而且凌晨的时候最多。
com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure
相信有一点经验的java开发对这个错误并不陌生，导致这个问题的原因也非常好定位，无非是客户端与数据库的连接被服务端主动断开，而客户端还傻傻的用这个已经被断开的连接去请求数据库，导致失败。
我们知道Mysql有一个wait_timeout的配置会自动将空闲时间超过这个值（通常为8小时）的连接断开，因为物理数据库的配置并没有变动，而且代码中连接池中的配置的最小空闲时间远小于数据库wait_timeout，所以这个假设首先被排除。
那么会不会是HAProxy将我们的连接给断开了呢(很有可能) ？ 于是我们查看了HAProxy的配置
timeout connect：60s // 定义haproxy将客户端请求转发至后端服务器所等待的超时时长
timeout client： 120s //客户端非活动状态的超时时长
timeout server： 120s //客户端与服务器端建立连接后，等待服务器端的超时时长
发现HAProxy会主动将空闲时间时间超过1分钟的连接断开，于是我们修改了druid的配置，将数据库空闲验证的时间修改为timeBetweenEvictionRunsMillis修改为50s（原来是60s,考虑到极限情况如果设置为60s的话依然会存在不能保活的情况），但是经过测试后我们发现问题依然存在。
既然将保活时间设置到了60s以内为什么还会出现连接被断开呢？稍微思考一下，可能的原因不外乎只有两个
- HAProxy维护的连接有问题
- 保活策略没有生效
顺着这个思路我们首先排除了HAProxy的问题（简单来说就是HAProxy会保持客户端和服务器的会话，保证客户端到HAProxy的连接和HAProxy到服务器的连接是一致的，他们的空闲时间始终是一样的） 那么我们再来看看是不是保活策略没有生效呢？我们目前用的是druid来管理我们的数据库连接池 ， 要弄清这个问题，我们得先看看druid是进行工作及怎么进行活性检测的。
image
上面这张图表示了druid在获取线程池的大致的逻辑过程：druid在初始化时会创建两个守护线程，分别承担线程的创建和销毁任务，当用户线程出现等待获取线程的操作时（且线程池中的线程数不大于最大活动线程数），创建线程会自动创建新的连接并放到线程池中，所以当用户线程需要新的连接时，只需要直接从线程池获取即可。用户线程从线程池中获取到连接会根据用户的配置决定是否线程进行有效性验证，如果验证线程有效则返回线程，如果无效则将该连接关闭，（DestoryConnectionThread自动回收已关闭的连接），然后尝试重新从连接池中获取连接，知道获取到有效连接为止并返回连接。下面我们来看看代码的具体实现。
public DruidPooledConnection getConnectionDirect(long maxWaitMillis) throws SQLException {
int notFullTimeoutRetryCnt = 0;
for (;;) { // 死循环，直到获取到有效连接为止，依赖CreateConnectionThread 保证 ，连接池中始终有有效连接资源
// handle notFullTimeoutRetry
DruidPooledConnection poolableConnection;
try {
poolableConnection = getConnectionInternal(maxWaitMillis); // 从连接池中获取连接
} catch (GetConnectionTimeoutException ex) {
if (notFullTimeoutRetryCnt <= this.notFullTimeoutRetryCount && !isFull()) {
notFullTimeoutRetryCnt++;
if (LOG.isWarnEnabled()) {
LOG.warn("not full timeout retry : " + notFullTimeoutRetryCnt);
}
continue;
}
throw ex;
}
if (isTestOnBorrow()) {
// 。。。这里不讨论 省略
} else {
Connection realConnection = poolableConnection.getConnection();
if (realConnection.isClosed()) {
discardConnection(null); // 传入null，避免重复关闭
continue; // 如果连接已经关闭，丢弃，尝试重新获取新的连接
}
if (isTestWhileIdle()) { // 验证空闲连接有效性的配置 testWhileIdle = true
final long currentTimeMillis = System.currentTimeMillis();
final long lastActiveTimeMillis = poolableConnection.getConnectionHolder().getLastActiveTimeMillis();
final long idleMillis = currentTimeMillis - lastActiveTimeMillis;
long timeBetweenEvictionRunsMillis = this.getTimeBetweenEvictionRunsMillis();
if (timeBetweenEvictionRunsMillis <= 0) {
timeBetweenEvictionRunsMillis = DEFAULT_TIME_BETWEEN_EVICTION_RUNS_MILLIS;
}
// 当前连接的空闲时间是否大于timeBetweenEvictionRunsMillis，如果大于才检测，否则跳过
if (idleMillis >= timeBetweenEvictionRunsMillis) {
boolean validate = testConnectionInternal(poolableConnection.getConnection());
if (!validate) {
if (LOG.isDebugEnabled()) {
LOG.debug("skip not validate connection.");
}
discardConnection(realConnection); //丢弃连接
continue; // 尝试重新获取新的连接
}
}
}
}
...
return poolableConnection;
}
}
protected boolean testConnectionInternal(Connection conn) {
...
try {
if (validConnectionChecker != null) {
// 检查连接有效性
return validConnectionChecker.isValidConnection(conn, validationQuery, validationQueryTimeout);
}
if (conn.isClosed()) {
return false;
}
if (null == validationQuery) {
return true;
}
Statement stmt = null;
ResultSet rset = null;
try {
stmt = conn.createStatement();
if (getValidationQueryTimeout() > 0) {
stmt.setQueryTimeout(validationQueryTimeout);
}
rset = stmt.executeQuery(validationQuery);
if (!rset.next()) {
return false;
}
} finally {
JdbcUtils.close(rset);
JdbcUtils.close(stmt);
}
return true;
} catch (Exception ex) {
// skip
return false;
} finally {
if (sqlFile != null) {
JdbcSqlStat.setContextSqlFile(sqlFile);
}
if (sqlName != null) {
JdbcSqlStat.setContextSqlName(sqlName);
}
}
}
public class MySqlValidConnectionChecker extends ValidConnectionCheckerAdapter implements ValidConnectionChecker, Serializable {
public static final int DEFAULT_VALIDATION_QUERY_TIMEOUT = 1000;
private static final long serialVersionUID = 1L;
private static final Log LOG = LogFactory.getLog(MySqlValidConnectionChecker.class);
private Class<?> clazz;
private Method ping;
// 是否使用ping 检测连接的有效性，如果为false 则使用select 1 查询
private boolean usePingMethod = false;
public MySqlValidConnectionChecker(){
try {
clazz = Utils.loadClass("com.mysql.jdbc.MySQLConnection");
if (clazz == null) {
clazz = Utils.loadClass("com.mysql.cj.jdbc.ConnectionImpl");
}
if (clazz != null) {
ping = clazz.getMethod("pingInternal", boolean.class, int.class);
}
if (ping != null) {
usePingMethod = true;
}
} catch (Exception e) {
LOG.warn("Cannot resolve com.mysql.jdbc.Connection.ping method. Will use 'SELECT 1' instead.", e);
}
configFromProperties(System.getProperties());
}
public boolean isValidConnection(Connection conn, String validateQuery, int validationQueryTimeout) throws Exception {
if (conn.isClosed()) {
return false;
}
if (usePingMethod) {
if (conn instanceof DruidPooledConnection) {
conn = ((DruidPooledConnection) conn).getConnection();
}
if (conn instanceof ConnectionProxy) {
conn = ((ConnectionProxy) conn).getRawObject();
}
// 当前的conn 是否是 com.mysql.jdbc.MySQLConnection （or com.mysql.cj.jdbc.ConnectionImpl）
if (clazz.isAssignableFrom(conn.getClass())) {
if (validationQueryTimeout < 0) {
validationQueryTimeout = DEFAULT_VALIDATION_QUERY_TIMEOUT;
}
//使用反射调用MySQLConnection.pingInternal 方法，检查连接有效性，并且会刷新连接的空闲时间
// 如果失败则会抛出异常，上层捕获
ping.invoke(conn, true, validationQueryTimeout * 1000);
return true;
}
}
// 当usePingMethod=false或者 conn 不是 com.mysql.jdbc.MySQLConnection （or com.mysql.cj.jdbc.ConnectionImpl） 会执行一下方法
if (validateQuery == null || validateQuery.length() == 0) {
return true;
}
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement();
if (validationQueryTimeout > 0) {
stmt.setQueryTimeout(validationQueryTimeout);
}
// 执行 select 1 ，并且会刷新连接的空闲时间
// 如果失败则会抛出异常，上层捕获
rs = stmt.executeQuery(validateQuery);
return true;
} finally {
JdbcUtils.close(rs);
JdbcUtils.close(stmt);
}
}
...
}
通过调试，我发现clazz.isAssignableFrom(conn.getClass())为false,也就是说这里的conn并不是com.mysql.jdbc.MySQLConnection，原来为了DB的读写分离项目使用的是数据库驱动是RepliationDriver而不是默认的Driver(jdbc:mysql:replication://), 因此使用的连接也是com.mysql.jdbc.ReplicationConnection, 而ReplicationConnection直接继承自 com.mysql.jdbc.Connection 并没有继承com.mysql.jdbc.MySQLConnection（仅限于mysql-connection-java 1.5.38版本之前，而我们线上使用的是1.5.35）
com.mysql.jdbc.ReplicationConnection在mysql-connector-java 1.5.38版本之前（ReplicationConnection源码）的是public class ReplicationConnection implements Connection, PingTarget
5.1.35版本1.5.38([GitHub源码]开始将
com.mysql.jdbc.ReplicationConnection抽象成接口（直接继承自com.mysql.jdbc.MySQLConnection），并使用他的的子类com.mysql.jdbc.JDBC4ReplicationMySQLConnection（JDBC4ReplicationMySQLConnection源码）来实现，com.mysql.jdbc.JDBC4ReplicationMySQLConnection的内部功能则是由新增的代理类com.mysql.jdbc.ReplicationConnectionProxy来实现的 （ReplicationConnectionProxy实现了原本com.mysql.jdbc.ReplicationConnection类实现的功能）
5.1.35版本
5.1.47版本
5.1.47版本
image
所以这里无论usePingMethod设置的值是什么，MySqlValidConnectionChecker都是执行SELECT 1 操作. 下面我们来具体看一下执行过程
先打开Mysql的日志 (当然是本地开发环境喽，我这里为了方便，主从库配置的是同一个地址)
set global general_log = on;
我们会得到以下日志（这里我只保留了核心日志信息，并进行了一些脱敏处理）
event_time thread_id command_type argument
2019-01-15T07:51:45.814624Z 18 Query SELECT 1 // 执行select 1的线程ID是 18
2019-01-15T07:51:45.822731Z 19 Query select * from xxx_table where xxx_column = 'xxx' limit 0,1 // 实际执行语句的线程ID是 19
我们很容易发现执行检查操作的线程和执行业务查询的线程是不一样的，因此我们可以断定检查用的连接和执行业务的数据库连接不是同一个，执行业务操作的数据库连接没有保活，空闲时间并没有被刷新，所以该连接一旦长时间没有访问就会被断开，导致出现连接不可用。
/**
* ！这里省略了非相关代码
* 简单来说就是维护了两个连接集合，一个是两个连接集合，一个包含主库连接，一个包含一个或多个从库连接， 当readonly = true 是使用从库连接，否则使用主库连接
*
*/
public class ReplicationConnection implements Connection, PingTarget {
protected Connection currentConnection;
protected LoadBalancedConnection masterConnection;
protected LoadBalancedConnection slavesConnection;
private ReplicationConnectionGroup connectionGroup;
private boolean readOnly = false;
/**
* 当设置readOnly时，切换当前连接
*/
public synchronized void setReadOnly(boolean readOnly) throws SQLException {
if (readOnly) {
if (this.currentConnection != this.slavesConnection) {
switchToSlavesConnection();
}
} else {
if (this.currentConnection != this.masterConnection) {
switchToMasterConnection();
}
}
this.readOnly = readOnly;
// allow master connection to be set to/from read-only if
// there are no slaves
if (this.currentConnection == this.masterConnection) {
this.currentConnection.setReadOnly(this.readOnly);
}
}
private synchronized void switchToMasterConnection() throws SQLException {
if (this.masterConnection == null || this.masterConnection.isClosed()) {
this.initializeMasterConnection();
}
swapConnections(this.masterConnection, this.slavesConnection);
this.masterConnection.setReadOnly(false);
}
private synchronized void switchToSlavesConnection() throws SQLException {
if (this.slavesConnection == null || this.slavesConnection.isClosed()) {
this.initializeSlaveConnection();
}
if (this.slavesConnection != null) {
swapConnections(this.slavesConnection, this.masterConnection);
this.slavesConnection.setReadOnly(true);
}
}
}
问题找到了，解决方案也比较简单， druid的高版本已经支持自定义ValidConnectionChecker
package com.hujiang.coupon.common.druid.check;
/**
* connection checker for Mysql Replication Driver (support Normal Driver as well)
*/
@Slf4j
public class MySqlReplicationValidConnectionChecker extends ValidConnectionCheckerAdapter implements ValidConnectionChecker, Serializable {
public static final int DEFAULT_VALIDATION_QUERY_TIMEOUT = 1;
public static final String DEFAULT_VALIDATION_QUERY = "SELECT 1";
private static final long serialVersionUID = 1L;
private boolean usePingMethod = false;
public MySqlReplicationValidConnectionChecker() {
configFromProperties(System.getProperties());
}
@Override
public void configFromProperties(Properties properties) {
String property = properties.getProperty("druid.mysql.usePingMethod");
if ("true".equals(property)) {
setUsePingMethod(true);
} else if ("false".equals(property)) {
setUsePingMethod(false);
}
}
public boolean isUsePingMethod() {
return usePingMethod;
}
public void setUsePingMethod(boolean usePingMethod) {
this.usePingMethod = usePingMethod;
}
@Override
public boolean isValidConnection(Connection conn, String validateQuery, int validationQueryTimeout) throws Exception {
if (conn.isClosed()) {
return false;
}
if (conn instanceof DruidPooledConnection) {
conn = ((DruidPooledConnection) conn).getConnection();
}
if (conn instanceof ConnectionProxy) {
conn = ((ConnectionProxy) conn).getRawObject();
}
if (usePingMethod) {
if (conn instanceof MySQLConnection) {
// 使用ping 方法替代原来的pingInternal 方法，ReplicationMySQLConnection内部实现中会同时执行master和slave节点的ping操作
((MySQLConnection) conn).ping();
return true;
}
}
String query = validateQuery;
if (validateQuery == null || validateQuery.isEmpty()) {
query = DEFAULT_VALIDATION_QUERY;
}
if(conn instanceof ReplicationConnection){
select_1(((ReplicationConnection) conn).getMasterConnection(),query,validationQueryTimeout);
select_1(((ReplicationConnection) conn).getSlavesConnection(),query,validationQueryTimeout);
}else{
select_1(conn, query, validationQueryTimeout);
}
return true;
}
private void select_1(Connection conn, String validateQuery, int validationQueryTimeout) throws SQLException {
Statement stmt = null;
ResultSet rs = null;
try {
stmt = conn.createStatement();
if (validationQueryTimeout > 0) {
stmt.setQueryTimeout(validationQueryTimeout);
}
rs = stmt.executeQuery(validateQuery);
} finally {
JdbcUtils.close(rs);
JdbcUtils.close(stmt);
}
}
}
// 可以看到com.mysql.jdbc.ReplicationMySQLConnection 的ping()方法的实现
public class ReplicationMySQLConnection extends MultiHostMySQLConnection implements ReplicationConnection {
@Override
public synchronized void ping() throws SQLException {
Connection conn;
try {
if ((conn = getValidatedMasterConnection()) != null) {
conn.ping();
}
} catch (SQLException e) {
if (isMasterConnection()) {
throw e;
}
}
try {
if ((conn = getValidatedSlavesConnection()) != null) {
conn.ping();
}
} catch (SQLException e) {
if (!isMasterConnection()) {
throw e;
}
}
}
...
}
网友评论