Java高并发数据库同步处理：高效任务调度与连接管理实践

import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

// 假设有一个数据库连接管理类
class Database {
    // 实际项目中应使用连接池
    public static Connection getConnection() throws SQLException {
        // 示例：从连接池获取连接
        // return databaseConnectionPool.getConnection();
        throw new UnsupportedOperationException("Implement connection pooling here.");
    }
}

/**
 * 封装单行数据库操作的任务
 */
public class DatabaseTask implements Runnable {
    private int databaseRowId; // 待处理的行ID

    public DatabaseTask(int rowId) {
        this.databaseRowId = rowId;
    }

    @Override
    public void run() {
        Connection connection = null;
        try {
            // 从连接池获取连接
            connection = Database.getConnection();
            // ... 对特定行 databaseRowId 进行数据读取、计算和更新/删除操作 ...
            System.out.println("Processing row: " + databaseRowId + " on thread: " + Thread.currentThread().getName());
            // 模拟耗时计算
            Thread.sleep(1500);
            // 更新行状态或删除行
            // 例如：UPDATE your_table SET status = 'CONSUMED' WHERE id = ?
            // 或 DELETE FROM your_table WHERE id = ?

        } catch (SQLException e) {
            System.err.println("Database error processing row " + databaseRowId + ": " + e.getMessage());
            // 适当的错误处理，例如记录日志、重试机制等
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重新设置中断状态
            System.err.println("Task for row " + databaseRowId + " was interrupted.");
        } finally {
            if (connection != null) {
                try {
                    connection.close(); // 将连接返回连接池
                } catch (SQLException e) {
                    System.err.println("Error closing connection: " + e.getMessage());
                }
            }
        }
    }
}

public class TaskScheduler {
    private final ExecutorService executor;
    private final Object databaseWriteLock = new Object(); // 示例：如果需要应用层面的全局写锁

    public TaskScheduler(int poolSize) {
        this.executor = Executors.newFixedThreadPool(poolSize);
    }

    /**
     * 提交数据库任务
     * @param rowId 待处理的行ID
     */
    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 static void main(String[] args) {
        TaskScheduler scheduler = new TaskScheduler(7); // 例如，7个并发线程

        // 模拟一个“工作发现者”组件，批量提交任务
        for (int i = 1; i <= 20; i++) { // 假设有20行数据需要处理
            scheduler.submitTask(i);
        }

        scheduler.shutdown();
    }
}

// 在实际项目中，HikariCP配置通常在应用启动时完成
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;

    static {
        HikariConfig config = new HikariConfig();
        config.setJdbcUrl("jdbc:mariadb://localhost:3306/your_database"); // 或 SQLite: jdbc:sqlite:your_database.db
        config.setUsername("your_user");
        config.setPassword("your_password");
        config.setMaximumPoolSize(10); // 根据并发量和数据库性能调整
        config.setMinimumIdle(5);
        config.setConnectionTimeout(30000); // 30 seconds
        config.setIdleTimeout(600000); // 10 minutes
        config.setMaxLifetime(1800000); // 30 minutes
        // 其他HikariCP配置...

        dataSource = new HikariDataSource(config);
    }

    public static Connection getConnection() throws SQLException {
        return dataSource.getConnection();
    }

    public static void closeDataSource() {
        if (dataSource != null) {
            dataSource.close();
        }
    }
}

// 在 DatabaseTask 的 run 方法内部
try (Connection connection = Database.getConnection()) {
    connection.setAutoCommit(false); // 开启事务

    // 1. 读取行数据
    // 例如：SELECT data_column FROM your_table WHERE id = ? FOR UPDATE; (悲观锁)
    // 或者 SELECT data_column, version FROM your_table WHERE id = ?; (乐观锁准备)

    // 2. 进行耗时计算
    // makeComputation(string);

    // 3. 更新/删除行
    // 例如：UPDATE your_table SET status = 'CONSUMED' WHERE id = ?;
    // 或 DELETE FROM your_table WHERE id = ?;

    connection.commit(); // 提交事务
} catch (SQLException e) {
    if (connection != null) {
        try {
            connection.rollback(); // 事务回滚
        } catch (SQLException ex) {
            System.err.println("Error rolling back transaction: " + ex.getMessage());
        }
    }
    System.err.println("Database error processing row " + databaseRowId + ": " + e.getMessage());
}