Java并发处理大规模数据库记录：优化与同步策略

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);
    }
}

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();
    }
}

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.");
        }
    }
}

// 假设这是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 类中
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
}