Oracle与Java：如何避免多线程读取百万级数据表重复行

针对百万级全表扫描场景，不依赖Java框架的前提下，可通过以下几种原生方案避免线程间读取重复数据：

1. 数据库行级锁+状态标记（通用兼容方案） #

在数据表中新增一个process_status字段（比如0=未处理，1=处理中，2=已完成），线程通过原子性的更新操作抢占数据行：

• 线程执行UPDATE语句锁定目标行并标记状态，仅当状态为未处理时才生效
• 根据更新影响行数判断是否成功抢占，成功则查询该行数据进行处理
• 处理完成后更新状态为已完成，异常时可重置为未处理

示例JDBC代码：

// 获取数据库连接（省略连接池/创建逻辑）
Connection conn = DriverManager.getConnection(url, user, pwd);
conn.setAutoCommit(false);

try {
    String updateSql = "UPDATE target_table SET process_status = 1 WHERE id = ? AND process_status = 0";
    PreparedStatement updateStmt = conn.prepareStatement(updateSql);
    updateStmt.setLong(1, targetId); // 可通过预查询获取待处理id列表，或循环尝试
    int affectedRows = updateStmt.executeUpdate();
    
    if (affectedRows > 0) {
        // 抢占成功，查询数据
        String selectSql = "SELECT * FROM target_table WHERE id = ?";
        PreparedStatement selectStmt = conn.prepareStatement(selectSql);
        selectStmt.setLong(1, targetId);
        ResultSet rs = selectStmt.executeQuery();
        if (rs.next()) {
            // 处理数据逻辑
            processData(rs);
            // 标记为已完成
            String finishSql = "UPDATE target_table SET process_status = 2 WHERE id = ?";
            PreparedStatement finishStmt = conn.prepareStatement(finishSql);
            finishStmt.setLong(1, targetId);
            finishStmt.executeUpdate();
        }
    }
    conn.commit();
} catch (SQLException e) {
    conn.rollback();
    // 异常处理，可重置状态或记录日志
} finally {
    conn.close();
}

2. 利用数据库跳过锁行特性（高效方案） #

如果使用MySQL 8.0+、PostgreSQL等支持SKIP LOCKED的数据库，可直接通过SELECT ... FOR UPDATE SKIP LOCKED语句获取未被锁定的行，无需额外状态字段：

• 线程每次查询一批未被锁定的数据行，自动跳过已被其他线程锁定的行
• 锁定的行在事务提交前不会被其他线程读取

示例JDBC代码：

Connection conn = DriverManager.getConnection(url, user, pwd);
conn.setAutoCommit(false);

try {
    // 每次获取100条未被锁定的行
    String selectSql = "SELECT * FROM target_table FOR UPDATE SKIP LOCKED LIMIT 100";
    PreparedStatement stmt = conn.prepareStatement(selectSql);
    ResultSet rs = stmt.executeQuery();
    
    while (rs.next()) {
        // 处理数据逻辑
        processData(rs);
    }
    conn.commit();
} catch (SQLException e) {
    conn.rollback();
} finally {
    conn.close();
}

优点：无需额外字段，性能更高；缺点：依赖数据库版本特性

3. 预分片处理（无锁方案） #

提前将全表数据按规则分片，每个线程处理固定分片区间，从根源避免重复读取：

• 先查询表中id的最大值和最小值，计算分片区间（比如按线程数均分）
• 每个线程仅处理自己区间内的数据，比如线程1处理id BETWEEN 1 AND 500000，线程2处理id BETWEEN 500001 AND 1000000
• 若id不连续，可改用哈希分片（比如id % thread_count = 0）

示例JDBC代码：

// 主线程计算分片区间
Connection conn = DriverManager.getConnection(url, user, pwd);
String countSql = "SELECT MIN(id), MAX(id) FROM target_table";
PreparedStatement countStmt = conn.prepareStatement(countSql);
ResultSet rs = countStmt.executeQuery();
rs.next();
long minId = rs.getLong(1);
long maxId = rs.getLong(2);
int threadNum = 2;
long range = (maxId - minId) / threadNum;

// 线程1处理区间
new Thread(() -> {
    try (Connection threadConn = DriverManager.getConnection(url, user, pwd)) {
        String sql = "SELECT * FROM target_table WHERE id >= ? AND id < ?";
        PreparedStatement stmt = threadConn.prepareStatement(sql);
        stmt.setLong(1, minId);
        stmt.setLong(2, minId + range);
        ResultSet threadRs = stmt.executeQuery();
        while (threadRs.next()) {
            processData(threadRs);
        }
    } catch (SQLException e) {
        e.printStackTrace();
    }
}).start();

// 线程2处理剩余区间
new Thread(() -> {
    try (Connection threadConn = DriverManager.getConnection(url, user, pwd)) {
        String sql = "SELECT * FROM target_table WHERE id >= ?";
        PreparedStatement stmt = threadConn.prepareStatement(sql);
        stmt.setLong(1, minId + range);
        ResultSet threadRs = stmt.executeQuery();
        while (threadRs.next()) {
            processData(threadRs);
        }
    } catch (SQLException e) {
        e.printStackTrace();
    }
}).start();

优点：无锁冲突，性能最优；缺点：分片规则需适配数据分布，若数据新增需重新分片

4. 原子偏移量分页（轻量方案） #

创建一个单独的计数器表，存储当前处理的偏移量，线程通过原子更新获取分页参数：

• 计数器表结构：CREATE TABLE process_counter (id INT PRIMARY KEY, current_offset BIGINT DEFAULT 0)
• 每个线程原子性地获取并增加偏移量，再用该偏移量做分页查询
• 注意：百万级数据用OFFSET性能较差，建议改用WHERE id > last_id的方式遍历

示例JDBC代码：

// 获取当前偏移量并原子增加
long getNextOffset(Connection conn) throws SQLException {
    String sql = "UPDATE process_counter SET current_offset = current_offset + 100 WHERE id = 1";
    PreparedStatement stmt = conn.prepareStatement(sql);
    stmt.executeUpdate();
    
    String selectSql = "SELECT current_offset FROM process_counter WHERE id = 1";
    PreparedStatement selectStmt = conn.prepareStatement(selectSql);
    ResultSet rs = selectStmt.executeQuery();
    rs.next();
    return rs.getLong(1) - 100; // 返回本次查询的起始偏移量
}

// 线程处理逻辑
new Thread(() -> {
    try (Connection conn = DriverManager.getConnection(url, user, pwd)) {
        conn.setAutoCommit(true);
        while (true) {
            long offset = getNextOffset(conn);
            String sql = "SELECT * FROM target_table LIMIT 100 OFFSET ?";
            PreparedStatement stmt = conn.prepareStatement(sql);
            stmt.setLong(1, offset);
            ResultSet rs = stmt.executeQuery();
            
            // 若无数据则退出循环
            if (!rs.next()) break;
            
            do {
                processData(rs);
            } while (rs.next());
        }
    } catch (SQLException e) {
        e.printStackTrace();
    }
}).start();

优点：实现简单；缺点：分页偏移量在百万级时性能下降明显，建议结合id范围优化

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