Java数据库同步咨询：200万行数据多线程并发处理与事务优化

一、SQLite并发核心配置 #

SQLite默认采用文件级锁，写操作会阻塞所有读写请求，要支撑你的并发需求，必须先开启WAL（Write-Ahead Logging）模式，这是SQLite实现高并发读写的基础：

-- 开启WAL模式
PRAGMA journal_mode=WAL;
-- 调整同步级别（如果可接受极低数据丢失风险，用NORMAL；否则保持FULL）
PRAGMA synchronous=NORMAL;
-- 设置缓存大小（例如20MB，提升内存命中率）
PRAGMA cache_size=-20000;

WAL模式下，读操作不会被写操作阻塞，多个写操作可以串行排队执行，能大幅提升并发吞吐量。

二、行级任务抢占的实现（避免重复处理） #

SQLite没有原生行级锁，我们可以通过原子性的状态标记实现“抢占式”任务分配，确保同一行不会被多个线程同时处理：

1. 给数据表新增status列，定义状态值：0=未处理、1=处理中、2=已处理、3=处理失败
2. 每个线程通过事务原子性地抢占一行未处理任务：

   BEGIN TRANSACTION;
   -- 原子抢占并标记为处理中，同时返回任务ID和数据
   UPDATE tasks 
   SET status = 1 
   WHERE id = (SELECT id FROM tasks WHERE status = 0 LIMIT 1) 
   RETURNING id, data_col;
   COMMIT;
   

   这条SQL会在事务内完成“查询未处理行→标记为处理中”的原子操作，其他线程无法再抢到同一行。

三、更新状态vs删除行的性能对比 #

• 更新状态：优先推荐。优势是保留历史数据便于排查，写操作是原地更新，磁盘IO开销小，不会产生数据库碎片。对于200万行的规模，状态更新的性能更稳定，后续查询未处理行的速度受碎片影响小。
• 删除行：仅适合不需要历史数据的场景。删除操作会产生磁盘碎片，随着数据量减少，前期查询速度会变快，但后期需要定期执行VACUUM整理碎片，而VACUUM会锁表，严重影响并发性能。

四、优化后的并发处理伪代码 #

修正原有逻辑，将数据库操作拆为短事务，避免计算过程持有锁：

ExecutorService executor = Executors.newFixedThreadPool(7);

// 初始化数据库配置（全局执行一次）
try (Connection conn = DriverManager.getConnection("jdbc:sqlite:tasks.db")) {
    conn.createStatement().execute("PRAGMA journal_mode=WAL;");
    conn.createStatement().execute("PRAGMA synchronous=NORMAL;");
    conn.createStatement().execute("PRAGMA cache_size=-20000;");
    // 给status列加索引，加速未处理行查询
    conn.createStatement().execute("CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);");
}

// 线程任务逻辑
executor.execute(() -> {
    while (true) {
        long taskId = -1;
        String data = null;

        // 第一步：抢占未处理任务
        try (Connection conn = DriverManager.getConnection("jdbc:sqlite:tasks.db")) {
            conn.setAutoCommit(false);
            String grabSql = "UPDATE tasks SET status = 1 WHERE id = (SELECT id FROM tasks WHERE status = 0 LIMIT 1) RETURNING id, data_col;";
            try (PreparedStatement pstmt = conn.prepareStatement(grabSql);
                 ResultSet rs = pstmt.executeQuery()) {
                if (!rs.next()) {
                    // 无未处理任务，退出线程
                    break;
                }
                taskId = rs.getLong("id");
                data = rs.getString("data_col");
            }
            conn.commit();
        } catch (SQLException e) {
            // 抢占失败，重试
            continue;
        }

        // 第二步：执行计算（不占用数据库锁）
        makeComputation(data);

        // 第三步：标记任务为已处理
        try (Connection conn = DriverManager.getConnection("jdbc:sqlite:tasks.db")) {
            conn.setAutoCommit(false);
            String finishSql = "UPDATE tasks SET status = 2 WHERE id = ?;";
            try (PreparedStatement pstmt = conn.prepareStatement(finishSql)) {
                pstmt.setLong(1, taskId);
                pstmt.executeUpdate();
            }
            conn.commit();
        } catch (SQLException e) {
            // 标记失败，设置为异常状态
            try (Connection conn = DriverManager.getConnection("jdbc:sqlite:tasks.db")) {
                conn.createStatement().execute(String.format("UPDATE tasks SET status = 3 WHERE id = %d", taskId));
            } catch (SQLException ex) {
                ex.printStackTrace();
            }
        }
    }
});

五、额外性能优化点 #

• 使用连接池：替换直接创建Connection的方式，用HikariCP等轻量连接池，减少连接创建销毁的开销。
• 避免长事务：所有数据库操作都拆为短事务，绝对不要把makeComputation这类耗时操作放在事务内，否则会导致锁持有时间过长，阻塞其他线程。
• 批量预处理：如果允许，可调整抢占逻辑为批量抓取多个任务（比如一次抢5行），减少数据库交互次数，但要注意批量操作的原子性。

内容的提问来源于stack exchange，提问作者janusz j