挑战与需求分析
在处理如200万行数据，且每行数据需要1-2秒的计算，并最终标记为已处理（删除或更新状态）的场景中，主要面临以下挑战：
- 并发访问与数据一致性： 多个线程需要同时读取未处理数据，并更新或删除已处理数据，必须避免竞态条件和脏读。
- 长时间计算与数据库锁定： 如果在计算期间持有数据库连接和行锁，将严重影响数据库的并发性能和吞吐量。
- 性能要求： 整体处理速度至关重要，需要高效利用系统资源。
- 资源管理： 频繁的数据库连接创建和关闭会带来显著开销。
为了解决这些问题，我们需要一个策略，将数据库操作与耗时计算解耦，并充分利用现代数据库的并发控制能力。
核心策略：任务分解与异步执行
将每个需要处理的数据库行视为一个独立的任务，并通过Java的ExecutorService进行异步调度和执行，是实现高并发的关键。
1. 任务封装：DatabaseTask
创建一个实现Runnable接口的DatabaseTask类，用于封装针对特定数据库行的处理逻辑。每个DatabaseTask实例负责处理一个或一组特定的数据库行。
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
public class DatabaseTask implements Runnable {
private int databaseRowId;
private String rowData; // 用于存储从数据库获取的数据
public DatabaseTask(int rowId) {
this.databaseRowId = rowId;
}
// 可选：如果任务在初始化时就能获取部分数据，可以这样构造
public DatabaseTask(int rowId, String data) {
this.databaseRowId = rowId;
this.rowData = data;
}
@Override
public void run() {
// 阶段1: 从数据库获取数据并标记为“处理中”
if (!fetchAndMarkProcessing()) {
System.err.println("Failed to fetch or mark row " + databaseRowId + " as processing.");
return;
}
// 阶段2: 执行耗时计算（不持有数据库连接）
System.out.println("Processing row " + databaseRowId + " with data: " + rowData);
try {
makeComputation(rowData); // 模拟耗时计算
Thread.sleep(1500); // 模拟1.5秒的计算时间
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.err.println("Computation interrupted for row " + databaseRowId);
// 考虑如何处理中断，例如标记为失败或重新排队
}
// 阶段3: 更新数据库状态为“已完成”或删除
if (!markAsConsumed()) {
System.err.println("Failed to mark row " + databaseRowId + " as consumed.");
// 考虑回滚或重试策略
} else {
System.out.println("Row " + databaseRowId + " successfully processed and marked as consumed.");
}
}
private boolean fetchAndMarkProcessing() {
try (Connection connection = Database.getConnection()) {
connection.setAutoCommit(false); // 开启事务
// 1. 锁定并读取行
String selectSql = "SELECT content FROM my_table WHERE id = ? AND status = 'NEW' FOR UPDATE";
try (PreparedStatement selectStmt = connection.prepareStatement(selectSql)) {
selectStmt.setInt(1, databaseRowId);
ResultSet rs = selectStmt.executeQuery();
if (rs.next()) {
this.rowData = rs.getString("content");
} else {
connection.rollback(); // 没有找到或已被处理
return false;
}
}
// 2. 标记为“处理中”
String updateSql = "UPDATE my_table SET status = 'PROCESSING' WHERE id = ?";
try (PreparedStatement updateStmt = connection.prepareStatement(updateSql)) {
updateStmt.setInt(1, databaseRowId);
updateStmt.executeUpdate();
}
connection.commit(); // 提交事务
return true;
} catch (SQLException e) {
System.err.println("Error fetching or marking row " + databaseRowId + " as processing: " + e.getMessage());
// 实际应用中应有更详细的日志和错误处理
return false;
}
}
private boolean markAsConsumed() {
try (Connection connection = Database.getConnection()) {
connection.setAutoCommit(false); // 开启事务
// 更新状态为 'CONSUMED' 或删除
String updateSql = "UPDATE my_table SET status = 'CONSUMED' WHERE id = ?"; // 推荐更新状态
// String deleteSql = "DELETE FROM my_table WHERE id = ?"; // 或删除
try (PreparedStatement updateStmt = connection.prepareStatement(updateSql)) {
updateStmt.setInt(1, databaseRowId);
updateStmt.executeUpdate();
}
connection.commit(); // 提交事务
return true;
} catch (SQLException e) {
System.err.println("Error marking row " + databaseRowId + " as consumed: " + e.getMessage());
// 实际应用中应有更详细的日志和错误处理
return false;
}
}
private void makeComputation(String data) {
// 模拟实际的业务计算逻辑
// System.out.println("Performing heavy computation for: " + data);
}
}2. 线程池管理：ExecutorService
使用ExecutorService来管理和执行DatabaseTask。根据系统资源（CPU核心数、数据库连接池大小等）合理配置线程池大小。
立即学习“Java免费学习笔记（深入）”；
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class TaskManager {
private static final int THREAD_POOL_SIZE = 7; // 根据实际情况调整
private ExecutorService executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
public void submitTask(int rowId) {
executor.submit(new DatabaseTask(rowId));
}
public void shutdown() {
executor.shutdown();
try {
if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}
}
// 示例：如何找到并提交任务
public void startProcessing() {
// 这是一个简化的示例，实际中应从数据库查询未处理的行ID
for (int i = 1; i <= 20; i++) { // 假设有20行数据需要处理
submitTask(i);
}
}
public static void main(String[] args) {
// 确保数据库连接池已初始化
Database.initConnectionPool();
TaskManager manager = new TaskManager();
manager.startProcessing();
manager.shutdown();
}
}数据库连接管理：连接池的重要性
频繁地创建和关闭数据库连接是性能瓶颈之一。使用数据库连接池是最佳实践，它预先创建并维护一定数量的数据库连接，供应用程序复用。
推荐：HikariCP
HikariCP 是目前Java领域性能最佳的连接池之一，配置简单且效率极高。
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import java.sql.Connection;
import java.sql.SQLException;
public class Database {
private static HikariDataSource dataSource;
// 数据库初始化方法，应在应用启动时调用一次
public static void initConnectionPool() {
HikariConfig config = new HikariConfig();
config.setJdbcUrl("jdbc:mariadb://localhost:3306/mydatabase"); // 或 jdbc:mysql, jdbc:sqlite
config.setUsername("user");
config.setPassword("password");
config.setMaximumPoolSize(20); // 根据并发线程数和数据库负载调整
config.setMinimumIdle(5);
config.setConnectionTimeout(30000); // 30 seconds
config.setIdleTimeout(600000); // 10 minutes
config.setMaxLifetime(1800000); // 30 minutes
// 针对特定数据库的优化，例如MariaDB/MySQL
config.addDataSourceProperty("cachePrepStmts", "true");
config.addDataSourceProperty("prepStmtCacheSize", "250");
config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
dataSource = new HikariDataSource(config);
System.out.println("HikariCP connection pool initialized.");
}
public static Connection getConnection() throws SQLException {
if (dataSource == null) {
throw new SQLException("Database connection pool not initialized. Call initConnectionPool() first.");
}
return dataSource.getConnection();
}
// 在应用关闭时关闭连接池
public static void closeConnectionPool() {
if (dataSource != null) {
dataSource.close();
System.out.println("HikariCP connection pool closed.");
}
}
}并发控制与事务管理：数据库层面的同步
对于行级并发控制和数据一致性，最可靠的机制是依赖底层数据库的事务和锁定功能。
1. 数据库选择
- 关系型数据库（如MariaDB/MySQL with InnoDB）： 强烈推荐使用支持事务和行级锁的数据库。InnoDB存储引擎提供了强大的事务支持（ACID特性）和行级锁定，能够有效处理高并发场景。
- SQLite： 虽然易于嵌入和使用，但SQLite在并发写入方面存在限制（默认是数据库级锁），对于高并发写入的场景可能不是最佳选择。
2. 两阶段数据库操作策略
为了避免在长时间计算期间锁定数据库，可以采用以下两阶段操作：
-
阶段一：获取并标记（短事务）
- 从连接池获取连接。
- 开启事务。
- 使用SELECT ... FOR UPDATE语句查询并锁定一条未处理的记录。
- 更新该记录的状态为“PROCESSING”（处理中）。
- 提交事务并释放连接。
- 将获取到的数据传递给DatabaseTask进行计算。
-
阶段二：更新或删除（短事务）
- 在计算完成后，从连接池获取新连接。
- 开启事务。
- 更新该记录的状态为“CONSUMED”（已处理）或直接删除该记录。
- 提交事务并释放连接。
这种策略确保了数据库连接和行锁只在必要的最短时间内被持有，最大程度地提高了并发性。
3. 标记为“已处理”的策略
- 更新状态列（推荐）： 在表中添加一个status列（例如：'NEW', 'PROCESSING', 'CONSUMED', 'FAILED'）。这种方法保留了历史数据，便于审计、回溯和错误处理。
- 删除行： 直接删除已处理的行。这种方法简单，但会丢失历史记录，不利于调试和数据恢复。
工作流编排与批处理
为了持续有效地处理200万行数据，需要一个“任务协调器”组件来不断地发现和提交新的DatabaseTask。
// 假设这是TaskCoordinator类
public class TaskCoordinator implements Runnable {
private ExecutorService executor;
private volatile boolean running = true;
private static final int BATCH_SIZE = 50; // 每次查询的行数
public TaskCoordinator(ExecutorService executor) {
this.executor = executor;
}
@Override
public void run() {
while (running && !Thread.currentThread().isInterrupted()) {
try {
// 查询未处理的行ID
// 注意：这里需要确保查询本身不会成为瓶颈，可以对status列建立索引
// 并且 LIMIT 子句在 FOR UPDATE 之前，以减少锁定范围
String selectNewRowsSql = "SELECT id FROM my_table WHERE status = 'NEW' ORDER BY id ASC LIMIT ?";
try (Connection connection = Database.getConnection();
PreparedStatement ps = connection.prepareStatement(selectNewRowsSql)) {
ps.setInt(1, BATCH_SIZE);
ResultSet rs = ps.executeQuery();
int tasksSubmitted = 0;
while (rs.next()) {
int rowId = rs.getInt("id");
executor.submit(new DatabaseTask(rowId));
tasksSubmitted++;
}
if (tasksSubmitted == 0) {
System.out.println("No new tasks found. Waiting...");
Thread.sleep(5000); // 如果没有新任务，等待一段时间再查询
} else {
System.out.println("Submitted " + tasksSubmitted + " new tasks.");
}
}
} catch (SQLException e) {
System.err.println("Error in TaskCoordinator querying new tasks: " + e.getMessage());
try {
Thread.sleep(10000); // 遇到数据库错误时等待更长时间
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println("TaskCoordinator interrupted.");
}
}
System.out.println("TaskCoordinator stopped.");
}
public void stop() {
this.running = false;
}
}在TaskManager中启动TaskCoordinator：
// ... 在 TaskManager 类中
private ExecutorService taskSubmitterExecutor = Executors.newSingleThreadExecutor();
private TaskCoordinator coordinator;
public void startProcessing() {
coordinator = new TaskCoordinator(executor); // executor 是处理任务的线程池
taskSubmitterExecutor.submit(coordinator); // 启动协调器
// ... 其他初始化
}
public void shutdown() {
coordinator.stop();
taskSubmitterExecutor.shutdown();
try {
if (!taskSubmitterExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
taskSubmitterExecutor.shutdownNow();
}
} catch (InterruptedException e) {
taskSubmitterExecutor.shutdownNow();
Thread.currentThread().interrupt();
}
// ... 原有的 executor shutdown
}注意事项与优化
- 错误处理与重试： 在DatabaseTask中，如果计算失败或数据库更新失败，应有完善的错误处理机制（如记录错误日志、将状态标记为FAILED、或实现指数退避重试）。
-
线程池大小调优： ExecutorService的线程池大小应根据CPU核心数、数据库连接池大小、I/O等待时间和任务类型（CPU密集型或I/O密集型）进行调整。
- 对于CPU密集型任务：N_CPU_CORES + 1
- 对于I/O密集型任务：N_CPU_CORES * (1 + WaitTime/CPUTime)
- 数据库索引： 确保status列和id列有合适的索引，以加速查询未处理记录和更新操作。
- 幂等性： 如果任务可能重试，确保makeComputation和数据库更新操作是幂等的，即多次执行相同操作不会产生额外副作用。
- 监控与日志： 实施详细的日志记录和性能监控，以便在生产环境中诊断问题和进行优化。
- **数据库事务隔离级别
