kafka数据同步，oracle数据库连接池

Question

服务且接收kafka消息，业务处理后，sql语句修改失败，服务器一直刷新： BEGIN check for expired resources.
FINISHED check for expired resources. [com.mchange.v2.resourcepool.BasicResourcePool@7f31a90b]
Checking for expired resources - Tue May 20 17:14:09 CST 2025 [com.mchange.v2.resourcepool.BasicResourcePool，重启服务会输出大量的同步数据之后就没有任何日

志输出。kafka消费消息的日志也不再输出，像是卡在哪里这是什么原因，怎么解决？

Answer 1

晚上好🌙🌙🌙
本答案参考通义千问

从你的描述来看，问题可能与以下几个方面有关：

1. Oracle 数据库连接池资源耗尽：

   • 日志中反复出现 BEGIN check for expired resources 和 FINISHED check for expired resources 的信息，表明数据库连接池可能存在资源泄漏或配置不当的问题。
   • 这些信息通常是由于数据库连接没有正确释放导致的。

2. Kafka 消息消费停滞：

   • Kafka 消费者日志不再输出，可能是由于消费者线程被阻塞或死锁，导致无法继续处理消息。

3. 服务重启后短暂恢复但最终停滞：

   • 服务重启后能够处理一部分数据，但最终又停滞，这可能与内存泄漏、连接池配置不足或数据库事务未正确提交有关。

解决方案

以下是解决问题的步骤和具体措施：

1. 检查数据库连接池配置

• 问题原因：连接池配置可能不正确，例如最大连接数设置过低、超时时间设置不合理等。

• 解决方案：

  • 增大数据库连接池的最大连接数（maxPoolSize）。
  • 调整连接池的空闲连接检查频率（idleConnectionTestPeriod）。
  • 确保数据库连接在使用后正确关闭，避免资源泄漏。

  修改代码示例：

  // 假设使用的是 HikariCP 或类似连接池
  HikariConfig config = new HikariConfig();
  config.setMaximumPoolSize(50); // 设置最大连接数
  config.setIdleTimeout(30000);  // 设置空闲连接超时时间为 30 秒
  config.setLeakDetectionThreshold(2000); // 设置泄漏检测阈值为 2 秒
  config.setConnectionTimeout(30000); // 设置连接超时时间为 30 秒
  

2. 检查 Kafka 消费器配置

• 问题原因：Kafka 消费者可能未正确配置，或者消费者线程被阻塞。

• 解决方案：

  • 确保 Kafka 消费者的 max.poll.records 配置合理，不要一次性拉取过多消息。
  • 设置合适的 session.timeout.ms 和 heartbeat.interval.ms，确保消费者能及时发送心跳。
  • 如果使用了手动提交偏移量，确保每次处理完消息后正确提交。

  修改代码示例：

  Properties props = new Properties();
  props.put("bootstrap.servers", "localhost:9092");
  props.put("group.id", "test-group");
  props.put("enable.auto.commit", "false"); // 手动提交偏移量
  props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  
  KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  consumer.subscribe(Arrays.asList("my-topic"));
  
  while (true) {
      ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
      for (ConsumerRecord<String, String> record : records) {
          processMessage(record.value()); // 处理消息
          consumer.commitSync(); // 手动提交偏移量
      }
  }
  

3. 监控和排查资源泄漏

• 问题原因：可能存在未关闭的数据库连接或资源泄漏。

• 解决方案：

  • 使用工具（如 JProfiler、VisualVM）监控内存使用情况，查找潜在的内存泄漏。
  • 在代码中增加日志，记录每次数据库连接的创建和关闭情况，确保每条 SQL 语句都能正确关闭连接。

  代码示例：

  try (Connection conn = dataSource.getConnection()) {
      try (PreparedStatement stmt = conn.prepareStatement("SELECT * FROM my_table")) {
          ResultSet rs = stmt.executeQuery();
          while (rs.next()) {
              processResultSet(rs);
          }
      }
  } catch (SQLException e) {
      logger.error("SQL error occurred", e);
  }
  

4. 增加日志和调试信息

• 问题原因：当前日志不足以定位问题的具体位置。

• 解决方案：

  • 在关键代码路径增加日志，特别是数据库操作和 Kafka 消费逻辑。
  • 记录每次处理的消息内容、SQL 语句执行结果以及异常信息。

  代码示例：

  logger.info("Processing message: {}", message);
  try {
      // 数据库操作
      processDatabaseQuery(query);
      logger.info("Database query executed successfully.");
  } catch (Exception e) {
      logger.error("Error processing database query", e);
  }
  

5. 检查数据库事务管理

• 问题原因：数据库事务未正确提交或回滚，可能导致死锁或资源占用。

• 解决方案：

  • 确保每个事务在完成后都正确提交或回滚。
  • 避免长时间持有数据库连接，尽量将事务范围缩小到最小。

  代码示例：

  try (Connection conn = dataSource.getConnection()) {
      conn.setAutoCommit(false); // 手动管理事务
      try (PreparedStatement stmt = conn.prepareStatement("UPDATE my_table SET column = ? WHERE id = ?")) {
          stmt.setString(1, newValue);
          stmt.setInt(2, id);
          stmt.executeUpdate();
          conn.commit(); // 提交事务
      } catch (SQLException e) {
          conn.rollback(); // 回滚事务
          throw e;
      }
  }
  

总结

通过以上步骤，可以逐步排查和解决数据库连接池资源耗尽、Kafka 消费器阻塞以及资源泄漏等问题。重点在于优化连接池配置、合理管理 Kafka 消费逻辑、监控资源使用情况，并确保事务管理正确无误。如果问题仍未解决，建议进一步分析日志和性能监控数据，定位具体的瓶颈点。