Java 多线程环境下数据库同步与并发处理：高效处理海量数据

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

public class DatabaseTask implements Runnable {
    private int databaseRowId;

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

    @Override
    public void run() {
        try (Connection connection = Database.getConnection()) {
            try {
                connection.setAutoCommit(false); // 开启事务

                // 1. 读取数据
                String data = readData(connection, databaseRowId);

                // 2. 执行计算
                String result = makeComputation(data);

                // 3. 更新数据库状态
                updateDatabase(connection, databaseRowId, result);

                connection.commit(); // 提交事务
            } catch (Exception e) {
                connection.rollback(); // 回滚事务
                e.printStackTrace();
            } finally {
                connection.setAutoCommit(true); // 恢复自动提交
            }

        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    private String readData(Connection connection, int rowId) throws SQLException {
        // 从数据库读取数据
        // 使用 PreparedStatement 避免 SQL 注入
        return "data from row " + rowId; // 模拟数据读取
    }

    private String makeComputation(String data) {
        // 模拟计算过程
        try {
            Thread.sleep(1000); // 模拟耗时计算
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return "computed result from " + data;
    }

    private void updateDatabase(Connection connection, int rowId, String result) throws SQLException {
        // 更新数据库状态
        // 使用 PreparedStatement 避免 SQL 注入
        System.out.println("Updated row " + rowId + " with result: " + result); // 模拟数据库更新
    }

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(7);

        for (int i = 1; i <= 20; i++) {
            executor.submit(new DatabaseTask(i)); // 提交任务到线程池
        }

        executor.shutdown(); // 关闭线程池
    }
}


class Database {
    private static HikariCPDataSource dataSource = new HikariCPDataSource();
    public static Connection getConnection() throws SQLException {
        return dataSource.getConnection();
    }
}

import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;

class HikariCPDataSource {

    private static HikariConfig config = new HikariConfig();
    private static HikariDataSource ds;

    HikariCPDataSource() {
        config.setJdbcUrl("jdbc:mysql://localhost:3306/your_database");
        config.setUsername("your_username");
        config.setPassword("your_password");
        config.setDriverClassName("com.mysql.cj.jdbc.Driver"); // 确保包含 MySQL JDBC 驱动
        config.addDataSourceProperty("cachePrepStmts", "true");
        config.addDataSourceProperty("prepStmtCacheSize", "250");
        config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
        ds = new HikariDataSource(config);
    }

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