Java 併發實戰：固定線程池批量處理 10 萬條優惠券數據與死鎖避坑指南

Java 併發實戰：固定線程池批量處理 10 萬條優惠券數據與死鎖避坑指南

業務場景：批量為 10 萬用戶發放活動優惠券，採用 10 線程固定池，每批處理 1000 條，含失敗重試機制。

任務類設計：實現 Callable 獲取返回值

public class InsertBatchThread implements Callable<Integer> {
    private int vdate, uid, count;
    private FundsInfoMapper fundsInfoMapper;
    private WmpsDayInterMapper wmpsDayInterMapper;
    private DataSource dataSource;

    // Spring 上下文懶加載 Mapper (雙重檢查鎖單例模式)
    public FundsInfoMapper getFundsInfoMapper() { /* ... */ }
    public WmpsDayInterMapper getWmpsDayInterMapper() { /* ... */ }

    public InsertBatchThread(int vdate, int uid, int count) {
        this.vdate = vdate; this.uid = uid; this.count = count;
        this.fundsInfoMapper = getFundsInfoMapper();
        this.wmpsDayInterMapper = getWmpsDayInterMapper();
    }

    @Override
    public Integer call() {
        int result = -1;
        try {
            // 1. 分頁查詢用戶
            List<Integer> listUsers = fundsInfoMapper.selectUserForInsertBatch(count * 1000, 1000);
            // 2. 批量構建優惠券數據
            List<WmpsDayInter> listOnePage = wmpsDayInterMapper.selectByInsertBatch(
                vdate, listUsers.get(0), listUsers.get(listUsers.size() - 1));
            // 3. JDBC 批量插入
            wmpsDayInterInsertBatch(listOnePage);
        } catch (Exception e) {
            result = count; // 返回失敗批次號用於重試
        }
        return result;
    }

    // JDBC 批量插入實現
    public int[] wmpsDayInterInsertBatch(List<WmpsDayInter> list) throws Exception {
        dataSource = SpringContextUtils.getBean("dataSource");
        PreparedStatement ps = dataSource.getConnection().prepareStatement(
            "INSERT IGNORE INTO t_a (uid, inter, cdate) VALUES (?,?,?)");
        for (WmpsDayInter obj : list) {
            ps.setInt(1, obj.getUid());
            ps.setBigDecimal(2, obj.getInter());
            ps.setBigDecimal(3, obj.getCdate());
            ps.addBatch();
        }
        return ps.executeBatch();
    }
}

業務協調類：分批提交、失敗收集、重試補償

@Service("FundsInfoService")
public class FundsInfoServiceImpl implements IFundsInfoService {
    @Override
    public void insertDayInter(Integer vdate, Integer uid) {
        // 參數校驗
        if (vdate == null || vdate <= 0 || uid == null || uid < 0) { /* log error */ return; }

        int totalNum = fundsInfoMapper.countForInsertBatchByUser();
        List<Integer> failedBatches = new ArrayList<>();

        try {
            ExecutorService pool = Executors.newFixedThreadPool(10);
            
            // 首輪提交
            for (int i = 0; i <= Math.floor(totalNum / 1000); i++) {
                Future<Integer> future = pool.submit(new InsertBatchThread(vdate, uid, i));
                if (future.get() >= 0) failedBatches.add(future.get()); // 收集失敗批次
            }

            // 重試補償機制
            if (!failedBatches.isEmpty()) {
                for (int batch : failedBatches) {
                    Future<Integer> retry = pool.submit(new InsertBatchThread(vdate, uid, batch));
                    if (retry.get() >= 0) failedBatches.add(retry.get()); // 遞歸重試邏輯可優化
                }
            }
            pool.shutdown();
        } catch (Exception e) { /* log error */ }
    }
}

核心避坑指南：數據分片維度決定死鎖風險

⚠️ 關鍵洞察：若分片維度設計不當（如按用戶 ID 範圍重疊），多線程同時「查詢-插入」同一數據行，極易產生死鎖——線程 A 持有鎖等待線程 B，線程 B 同樣等待 A。

安全分片原則

• 互斥分片：每批次處理數據集合完全不重疊 (如 LIMIT offset, pageSize)
• 冪等設計：SQL 使用 INSERT IGNORE 或 ON DUPLICATE KEY UPDATE
• 鎖粒度控制：避免長事務，批量操作盡量自動提交
• 重試幂等：重試邏輯必須冪等，防止重複發券

生產級優化建議

1. 使用 ThreadPoolExecutor 替代 Executors，配置有界隊列與拒絕策略
2. 引入 CompletableFuture 實現非阻塞編排，替代 future.get() 同步等待
3. 批次失敗記錄持久化，支援人工介入與斷點續傳
4. 監控指標：任務耗時分佈、隊列積壓、拒絕次數、重試成功率

此案例展示了從「能跑通」到「穩健生產」的併發工程化思維：分片策略決定並發安全，補償機制保證最終一致性。