您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 併發任務處理實戰：雙層線程池協同、異步查詢聚合與緩存穿透防護

編輯：Java 相關知識

生產級 Java 併發任務處理模式：從線程池設計到緩存一致性

面對海量數據聚合查詢 (如按日期、分區統計創角數)，單線程串行耗時不可接受。本文展示雙層線程池協同、CompletableFuture 編排、緩存旁路模式的工程化實現。

業務場景建模

查詢參數：起始日期、結束日期、渠道
數據分片：30 個遊戲大區
輸出：每日每區創角總數 Map<Date, Long>
非功能：秒級響應、支援高併發、緩存命中率優化

架構設計：雙層線程池隔離

// 1. 業務線程池 (CPU 密集：聚合、序列化、緩存操作)
ThreadPoolExecutor bizPool = new ThreadPoolExecutor(
    50, 200, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(1000),
    namedThreadFactory("Biz-Agg-"),
    new ThreadPoolExecutor.CallerRunsPolicy() // 降級：調用者執行
);

// 2. IO 線程池 (IO 密集：JDBC 查詢、Redis 訪問、HTTP 調用)
ThreadPoolExecutor ioPool = new ThreadPoolExecutor(
    50, 50, Long.MAX_VALUE, TimeUnit.SECONDS,
    new LinkedTransferQueue<>(), // 無界轉移隊列，生產者消費者直連
    namedThreadFactory("IO-JDBC-"),
    new ThreadPoolExecutor.AbortPolicy()
);

核心流程實現

public Map countCreateRole(String fdate, String tdate, String channel) {
    // 1. 日期範圍展開
    Set dates = JdbcTool.getDateRangeByDay(fdate, tdate, "yyyy-MM-dd");
    List zones = gameConfig.getZones(); // 30 個大區
    
    // 2. 外層：按日期並行 (業務線程池)
    List> dateFutures = dates.stream().map(date -> 
        CompletableFuture.runAsync(() -> {
            AtomicLong dayTotal = new AtomicLong();
            
            // 3. 內層：按大區並行 (IO 線程池)
            List> zoneFutures = zones.stream().map(zone -> 
                CompletableFuture.runAsync(() -> {
                    // 4. 緩存旁路模式
                    String cacheKey = "role:create:" + zone + ":" + date + ":" + channel;
                    String cached = redisTemplate.opsForValue().get(cacheKey);
                    long count;
                    if (cached != null) {
                        count = Long.parseLong(cached);
                    } else {
                        // 5. 兜底 DB 查詢
                        count = jdbcTemplate.queryForObject(
                            "SELECT COUNT(*) FROM " + zone + "_log.role " +
                            "WHERE DATE(create_time)=? AND channel_id=?", 
                            Long.class, date, channel);
                        // 6. 回寫緩存 (TTL 24h)
                        redisTemplate.opsForValue().set(cacheKey, String.valueOf(count), 24, TimeUnit.HOURS);
                    }
                    dayTotal.addAndGet(count);
                }, ioPool)
            ).collect(Collectors.toList());
            
            // 7. 等待所有大區完成
            CompletableFuture.allOf(zoneFutures.toArray(new CompletableFuture[0])).join();
            
            // 8. 線程安全寫入結果
            synchronized (result) {
                result.put(date, dayTotal.get());
            }
        }, bizPool)
    ).collect(Collectors.toList());
    
    // 9. 等待所有日期完成
    CompletableFuture.allOf(dateFutures.toArray(new CompletableFuture[0])).join();
    return result;
}

關鍵優化點深度解析

1. LinkedTransferQueue 零拷貝傳遞

IO 線程池使用 `LinkedTransferQueue`，任務提交者直接傳遞給消費者線程，無隊列緩衝開銷，適合任務提交速度 ≥ 處理速度的場景。

2. 緩存鍵設計與穿透防護

// 鍵結構：業務:操作:維度1:維度2...
String cacheKey = "role:create:" + zone + ":" + date + ":" + channel;

// 防穿透：空值緩存 (短 TTL)
if (count == 0) {
    redisTemplate.opsForValue().set(cacheKey, "0", 5, TimeUnit.MINUTES);
}

// 防雪崩：TTL 加隨機抖動
long ttl = 24 * 3600 + ThreadLocalRandom.current().nextInt(3600);

3. 異常傳播與熔斷

CompletableFuture.runAsync(() -> {
    try { /* 查詢邏輯 */ } 
    catch (Exception e) {
        log.error("Zone {} query failed", zone, e);
        // 可引入 Resilience4j 熔斷器
        throw new CompletionException(e); // 保留堆疊傳播
    }
}, ioPool).exceptionally(ex -> {
    metrics.increment("query.failed");
    return null; // 降級返回 null，上層兜底
});

4. 優雅關閉與資源釋放

@PreDestroy
public void shutdown() {
    // 1. 停止接受新任務
    bizPool.shutdown();
    ioPool.shutdown();
    
    // 2. 等待現有任務完成 (最長 2 分鐘)
    try {
        if (!bizPool.awaitTermination(120, TimeUnit.SECONDS)) {
            bizPool.shutdownNow(); // 強制中斷
        }
        if (!ioPool.awaitTermination(120, TimeUnit.SECONDS)) {
            ioPool.shutdownNow();
        }
    } catch (InterruptedException e) {
        bizPool.shutdownNow();
        ioPool.shutdownNow();
        Thread.currentThread().interrupt();
    }
}