问题描述

报错信息

在数据收集阶段，flume自定义source会通过查询本地mysql库解析用户ip地址；运行一段时间后，会出现如下报错，flume重启（重新建立数据库连接池）后又能正常一段时间
druid-1.1.0版本
mysql-connector-5.1.35

  1. catch getAddressOfIp error JSON:
  2. {"@type":"java.sql.SQLException",
  3. "cause":{"@type":"com.mysql.jdbc.exceptions.jdbc4.CommunicationsException",
  4. "cause":{"@type":"java.io.EOFException","localizedMessage":"Can not read response from server. Expected to read 4 bytes, read 0 bytes before connection was unexpectedly lost.","message":"Can not read response from server. Expected to read 4 bytes, read 0 bytes before connection was unexpectedly lost.",
  5. "stackTrace":
  6. {"className":"com.mysql.jdbc.MysqlIO","lineNumber":2914,"methodName":"readFully"},
  7. {"className":"com.mysql.jdbc.MysqlIO","lineNumber":3332,"methodName":"reuseAndReadPacket"},
  8. {"className":"com.mysql.jdbc.MysqlIO","lineNumber":3322,"methodName":"reuseAndReadPacket"},
  9. {"className":"com.mysql.jdbc.MysqlIO","lineNumber":3762,"methodName":"checkErrorPacket"},
  10. {"className":"com.mysql.jdbc.MysqlIO","lineNumber":2435,"methodName":"sendCommand"},
  11. {"className":"com.mysql.jdbc.MysqlIO","lineNumber":2582,"methodName":"sqlQueryDirect"},
  12. {"className":"com.mysql.jdbc.ConnectionImpl","lineNumber":2535,"methodName":"execSQL"},
  13. {"className":"com.mysql.jdbc.PreparedStatement","lineNumber":1911,"methodName":"executeInternal"},
  14. {"className":"com.mysql.jdbc.PreparedStatement","lineNumber":2034,"methodName":"executeQuery"},
  15. {"className":"com.alibaba.druid.pool.DruidPooledPreparedStatement","lineNumber":228,"methodName":"executeQuery"},
  16. {"className":"net.xinhuamm.sinks.MySinkKafka","lineNumber":245,"methodName":"getAddressOfIp"},
  17. {"className":"net.xinhuamm.sinks.MySinkKafka","lineNumber":119,"methodName":"process"}

我的代码

  1. public class DruidManager {
  2.     private DruidManager() {
  3.     }
  5.     private static DruidManager single = null;
  6.     private DruidDataSource dataSource;
  8.     public synchronized static DruidManager getInstance() {
  9.         if (single == null) {
  10.             single = new DruidManager();
  11.             single.initPool();
  12.         }
  13.         return single;
  14.     }
  16.     private void initPool() {
  17.         dataSource = new DruidDataSource();
  18.         dataSource.setDriverClassName("com.mysql.jdbc.Driver");
  19.         dataSource.setUrl("jdbc:mysql://hadoop-master:3306/noah_xinhua?characterEncoding=UTF-8");
  20.         dataSource.setUsername("noah_dba");
  21.         dataSource.setPassword("xinhua098");
  23.         dataSource.setInitialSize(50);
  24.         dataSource.setMinIdle(20);
  25.         //maxIdle值与maxActive值应配置的接近.当连接数超过maxIdle值后，刚刚使用完的连接（刚刚空闲下来）会立即被销毁。而不是想要的空闲M秒后再销毁起一个缓冲作用。若maxIdle与maxActive相差较大，在高负载的系统中会导致频繁的创建、销毁连接，连接数在maxIdle与maxActive间快速频繁波动，这不是想要的。高负载系统的maxIdle值可以设置为与maxActive相同或设置为-1(-1表示不限制)，让连接数量在minIdle与maxIdle间缓冲慢速波动。
  26.         //最大空闲连接，连接池中最大的空闲的连接数，超过的空闲连接将被释放，如果设置为负数表示不限制（默认为8个，maxIdle不能设置太小，因为假如在高负载的情况下，连接的打开时间比关闭的时间快，会引起连接池中idle的个数上升超过maxIdle，而造成频繁的连接销毁和创建，类似于jvm参数中的Xmx设置)
  27. //        dataSource.setMaxIdle(800);
  28.         dataSource.setMaxActive(800); //连接池的最大数据库连接数。设为0表示无限制。 一般把maxActive设置成可能的并发量就行了
  29.         dataSource.setMaxWait(5000);//从池中取连接的最大等待时间，单位ms.当没有可用连接时，连接池等待连接释放的最大时间，超过该时间限制会抛出异常，如果设置-1表示无限等待（默认为无限）
  31.         dataSource.setPoolPreparedStatements(true);
  32.         dataSource.setMaxPoolPreparedStatementPerConnectionSize(10);
  33.         //每30秒运行一次空闲连接回收器，配置timeBetweenEvictionRunsMillis = "30000"后，每30秒运行一次空闲连接回收器（独立线程）。并每次检查3个连接，如果连接空闲时间超过30分钟就销毁。销毁连接后，连接数量就少了，如果小于minIdle数量，就新建连接，维护数量不少于minIdle，过行了新老更替。
  34.         dataSource.setTimeBetweenEvictionRunsMillis(60000); //DBCP会启用独立的工作线程300秒定时检查
  35.         dataSource.setMinEvictableIdleTimeMillis(300000); //池中的连接空闲5分钟后被回收
  38.         dataSource.setTestWhileIdle(true);
  39.         dataSource.setTestOnBorrow(true); // <!-- 每次获取一个连接的时候，验证一下连接是否可用，语句在validationQuery里面 -->
  40.         dataSource.setTestOnReturn(false);
  41.         dataSource.setValidationQuery("SELECT 1"); //<!-- 数据库连接可用性测试语句 -->
  42. //        dataSource.setRemoveAbandoned(true); //<!-- 自动处理连接未关闭的问题，Setting this to true can recover db connections from poorly written applications which fail to close a connection.  -->
  43. //        dataSource.setRemoveAbandonedTimeout(180); //<!-- 连接使用后1分钟未关闭，则抛弃 -->
  44.     }
  46.     //要考虑多线程的情况
  47.     public Connection getConnection() {
  48.         Connection connection = null;
  49.         try {
  50.             synchronized (dataSource) {
  51.                 connection = dataSource.getConnection();
  52.             }
  53.         } catch (SQLException e) {
  54.             e.printStackTrace();
  55.         }
  56.         return connection;
  57.     }
  58. }
  1. public class MySinkKafka extends AbstractSink implements Configurable {
  3.     private static final Logger LOG = LoggerFactory.getLogger(MySinkKafka.class);
  5.     KafkaProducer producer;
  6.     Connection conn;
  7.     private JedisUtil jedisUtil;
  9.     private Boolean enableLog;
  12.     @Override
  13.     public void start() {
  14.         producer = getProducer();
  16.         DruidManager druidManager = DruidManager.getInstance();
  17.         conn = druidManager.getConnection();
  19.         jedisUtil =  JedisUtil.getInstance();
  20.     }
  22.     @Override
  23.     public void stop () {
  24.         // Disconnect from the external respository and do any
  25.         // additional cleanup (e.g. releasing resources or nulling-out
  26.         // field values) ..
  27.         System.out.println("MySinkKafka stop()------->");
  28.         LOG.info("MySinkKafka stop()------->");
  30.         if (producer != null) {
  31.             producer.close();
  32.         }
  34.         if (conn != null) {
  35.             try {
  36.                 conn.close();
  37.             } catch (SQLException e) {
  38.                 e.printStackTrace();
  39.             }
  40.         }
  41.     }
  43.     @Override
  44.     public Status process() {
  45.         ...
  46.     }
  48.     private static void assembleMap(JSONObject container, JSONObject jsonObject) {
  49.         ...
  50.     }
  52.     /**
  53.      * 获取生产者
  54.      *
  55.      * @author liujiapeng
  56.      * @date 2018/10/9 9:42
  57.      * @returnType org.apache.kafka.clients.producer.KafkaProducer
  58.      */
  59.     private KafkaProducer getProducer() {
  60.         ...
  61.     }
  63.     @Override
  64.     public void configure(Context context) {
  65.         ...
  66.     }
  68.     /**
  69.      * 使用淘宝的开放接口解析IP
  70.      *
  71.      * @param ip
  72.      * @return
  73.      */
  74.     private AddressInfo parsingIP(String ip) {
  75.         ...
  76.     }
  78.     /**
  79.      * mysql查询
  80.      * @param ip
  81.      * @return
  82.      */
  83.     private AddressInfo getAddressOfIp(String ip) throws SQLException {
  84.         String key = "mysql_" + ip;
  85.         String address = jedisUtil.get(key);
  86.         LOG.info("getAddressOfIp: key-" + key + ",address-" + address);
  87.         if (StringUtils.isEmpty(address) || StringUtils.equals("{}", address)) {
  88.             AddressInfo addressInfo = new AddressInfo();
  90.             PreparedStatement ps = null;
  91.             ResultSet rs = null;
  92.             try {
  93.                 String sql = "select country, province, city from ip where INET_ATON('" + ip + "') BETWEEN ip_start_num AND ip_end_num LIMIT 1";
  94.                 LOG.info("getAddressOfIp: sql:" + sql);
  95.                 ps = conn.prepareStatement(sql);
  96.                 rs = ps.executeQuery();
  97.                 while (rs.next()) {
  98.                     addressInfo.setCountry(rs.getString(1));
  99.                     addressInfo.setProvince(rs.getString(2));
  100.                     addressInfo.setCity(rs.getString(3));
  101.                 }
  102.             } catch (Exception e) {
  103.                 LOG.error("catch getAddressOfIp error:" + e.getMessage());
  104.                 LOG.error("catch getAddressOfIp error JSON:" + JSON.toJSONString(e));
  105.                 e.printStackTrace();
  106.             } finally {
  107.                 if (ps != null) {
  108.                     ps.close();
  109.                 }
  110.                 if (rs != null) {
  111.                     rs.close();
  112.                 }
  113.             }
  115.             if (StringUtils.isNotEmpty(addressInfo.getProvince())) {
  116.                 LOG.info("result:" + JSON.toJSONString(addressInfo));
  117.             }
  119.             return addressInfo;
  120.         }
  121.         return JSON.parseObject(address, AddressInfo.class);
  122.     }
  123. }

懵逼的摸索

first of all，打开度娘搜一搜~，各种修改连接池配置，以为找到银弹了，很兴奋的一个个试了过来，发现还是不行（吐槽下这个bug真难复现，本地还想不到调试的办法）；
只能硬着头皮装模做样瞅瞅源码咯~，简化了下代码，报错出在rs = ps.executeQuery() ，跟踪里面的代码没看出啥问题（主要还是自己菜，看懂不多），最下面的异常在MysqlIO类检查异常方法checkErrorPacket，发送连接拿到的inputStream长度为0导致的，报错是catch getAddressOfIp error:connection holder is null （最后的异常层层外抛也没找到这个异常）。看不下去了，我还是想直接点，找holder！虽然异常是代码爆出来的，但是，这个PreparedStatement是上面初始化的，感觉还是这个鬼有问题；往里一翻，又懵逼了，草草翻了下，大概了解了为啥度娘搜到的解决方案是改配置了，直接看拿到的连接DruidDataSource.getConnection()，看到了一堆数据库连接配置，根据代码的逻辑进行配置，试了下还是不行(其实根据代码逻辑配置和度娘上面差不多得，笑~)；咋办，脑壳疼。
最最最后，抱着一丝希望，希望他是个bbbbbug，去github上搜了下issue，发现遇到这个问题的人不少，看了下也没咋看懂，但的确有相关的一个bug不知道是不是我踩的这个。修复版本是druid-1.1.15，我就换成了次新版本druid-1.1.22，后续看试验情况等bug复现吧（希望别出来啊~）

观察了两天，emmmm… 还是没有解决（无能怒吼），但是好消息是报错换了

  1. java.sql.SQLException: connection disabled
  2.     at com.alibaba.druid.pool.DruidPooledConnection.checkStateInternal(DruidPooledConnection.java:1169)
  3.     at com.alibaba.druid.pool.DruidPooledConnection.checkState(DruidPooledConnection.java:1154)
  4.     at com.alibaba.druid.pool.DruidPooledConnection.prepareStatement(DruidPooledConnection.java:337)
  5.     at net.xinhuamm.sinks.MySinkKafka.getAddressOfIp(MySinkKafka.java:246)
  6.     at net.xinhuamm.sinks.MySinkKafka.process(MySinkKafka.java:121)
  7.     at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:67)
  8.     at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:145)
  9.     at java.lang.Thread.run(Thread.java:748)
  10. Caused by: com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: The last packet successfully received from the server was 31,749,415 milliseconds ago.  The last packet sent successfully to the server was 31,749,971 milliseconds ago. is longer than the server configured value of 'wait_timeout'. You should consider either expiring and/or testing connection validity before use in your application, increasing the server configured values for client timeouts, or using the Connector/J connection property 'autoReconnect=true' to avoid this problem.
  11.     at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
  12.     at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
  13.     at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
  14.     at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
  15.     at com.mysql.jdbc.Util.handleNewInstance(Util.java:389)
  16.     at com.mysql.jdbc.SQLError.createCommunicationsException(SQLError.java:1038)
  17.     at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3609)
  18.     at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2417)
  19.     at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2582)
  20.     at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2535)
  21.     at com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement.java:1911)
  22.     at com.mysql.jdbc.PreparedStatement.executeQuery(PreparedStatement.java:2034)
  23.     at com.alibaba.druid.pool.DruidPooledPreparedStatement.executeQuery(DruidPooledPreparedStatement.java:227)
  24.     at net.xinhuamm.sinks.MySinkKafka.getAddressOfIp(MySinkKafka.java:247)
  25.     ... 4 more
  26. Caused by: java.net.SocketException: Broken pipe (Write failed)
  27.     at java.net.SocketOutputStream.socketWrite0(Native Method)
  28.     at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:111)
  29.     at java.net.SocketOutputStream.write(SocketOutputStream.java:155)
  30.     at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
  31.     at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140)
  32.     at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3591)
  33.     ... 11 more

继续查呗，换个思路，我去server端也就是mysql日志里找找。
此处推荐下这篇文章，简单清晰：MySQL令人头疼的Aborted告警案例分析
cat /var/log/mysql/error.log

  1. 2020-09-24T03:30:38.097616Z 15839 [Note] Aborted connection 15839 to db: 'azkaban' user: 'root' host: 'hadoop-slave1' (Got timeout reading communication packets)
  2. 2020-09-24T03:30:38.127735Z 15840 [Note] Aborted connection 15840 to db: 'azkaban' user: 'root' host: 'hadoop-slave1' (Got timeout reading communication packets)
  3. 2020-09-24T03:37:41.710451Z 15841 [Note] Aborted connection 15841 to db: 'azkaban' user: 'root' host: 'hadoop-slave1' (Got timeout reading communication packets)
  4. 2020-09-24T09:27:48.004925Z 16285 [Note] Aborted connection 16285 to db: 'noah_xinhua' user: 'root' host: '115.236.34.2' (Got an error reading communication packets)
  5. 2020-09-24T09:54:27.068765Z 16286 [Note] Aborted connection 16286 to db: 'azkaban' user: 'root' host: '115.236.34.2' (Got an error reading communication packets)
  6. 2020-09-24T09:54:27.068782Z 16287 [Note] Aborted connection 16287 to db: 'azkaban' user: 'root' host: '115.236.34.2' (Got an error reading communication packets)

这错误信息对不上啊…emmm郁闷了，115.236.34.2（我的网络ip），我预期是能看到hadoop-slave2的host连接报错信息。gg 这日志没价值
看了下mysql配置，max_allowed_packet、interactive_timeout和wait_timeout 正常；

回头想了下，wait_timeout 是28800（即8小时），生产环境理论上不会出现8小时没用到连接，但是测试环境的确更容易复现出这个bug，所需要的时间更短，所以我先将等待时长设置成2628000（即1个月），坐等再观察观察…

如果bug还存在，考虑替换掉连接池，确保是否是db.close()调用问题；

还是gg了，调整mysql连接配置失败，我对conn加了行conn.close()先试试，如果不行，换调druid为dbcp。