引言

在現(xiàn)代軟件的開(kāi)發(fā)中，數(shù)據(jù)處理規(guī)模超過(guò)10w的情況已是常態(tài)，無(wú)論是電商平臺(tái)訂單批處理、物聯(lián)網(wǎng)設(shè)備日志分析等，無(wú)處不在挑戰(zhàn)數(shù)據(jù)處理的極限，并且，如何高效處理10w條數(shù)據(jù)也已成為如今常見(jiàn)但極具挑戰(zhàn)性的場(chǎng)景面試題，本文將闡釋這一挑戰(zhàn)的核心與架構(gòu)思路。

一.核心需求與挑戰(zhàn)

實(shí)際應(yīng)用場(chǎng)景

• 日志處理系統(tǒng)：?jiǎn)稳债a(chǎn)生10萬(wàn)+條用戶(hù)操作日志，需在1小時(shí)內(nèi)完成清洗、聚合、存儲(chǔ)
• 批量計(jì)算任務(wù)：每月用戶(hù)賬單生成，涉及10萬(wàn)用戶(hù)的數(shù)據(jù)計(jì)算
• 實(shí)時(shí)分析系統(tǒng)：廣告點(diǎn)擊流實(shí)時(shí)分析，每秒處理千條數(shù)據(jù)
• 數(shù)據(jù)同步作業(yè)：跨系統(tǒng)數(shù)據(jù)遷移，單表10萬(wàn)記錄同步

核心性能瓶頸

• CPU瓶頸：復(fù)雜計(jì)算、序列化/反序列化
•  內(nèi)存瓶頸：大對(duì)象持有、內(nèi)存泄漏
•  I/O瓶頸：數(shù)據(jù)庫(kù)查詢(xún)、文件讀寫(xiě)
•  網(wǎng)絡(luò)瓶頸：跨服務(wù)調(diào)用、帶寬限制

想讓程序一次性執(zhí)行10w條數(shù)據(jù)肯定是不可能的，所以我們必然要對(duì)整個(gè)過(guò)程進(jìn)行優(yōu)化，一般來(lái)說(shuō)，我們可以先從內(nèi)存和時(shí)間復(fù)雜度的角度去考慮問(wèn)題。

二.選擇高效的數(shù)據(jù)結(jié)構(gòu)

首先是數(shù)據(jù)的存儲(chǔ)，我們可以選擇更加高效的數(shù)據(jù)結(jié)構(gòu)，如查詢(xún)復(fù)雜度O(1)的Map，接下來(lái)我們從時(shí)間復(fù)雜度和內(nèi)存優(yōu)化分別來(lái)解決數(shù)據(jù)存儲(chǔ)時(shí)的高額消費(fèi)。

時(shí)間復(fù)雜度優(yōu)化

我們一般使用List去存儲(chǔ)數(shù)據(jù)，不過(guò)這樣會(huì)導(dǎo)致一個(gè)問(wèn)題就是查詢(xún)的時(shí)候復(fù)雜度達(dá)到了O(n)的時(shí)間復(fù)雜度，并且處理數(shù)據(jù)不一定要求有序，我們何不使用Map去存儲(chǔ)數(shù)據(jù)。

// 不推薦：O(n)查找
List<User> userList = new ArrayList<>();
User findUser = userList.stream()
    .filter(u -> u.getId().equals(targetId))
    .findFirst()
    .orElse(null);
// 推薦：O(1)查找
Map<String, User> userMap = new HashMap<>(100000 * 4/3 + 1); // 預(yù)分配
User findUser = userMap.get(targetId);

內(nèi)存優(yōu)化

• 預(yù)分配內(nèi)存：通過(guò)在new對(duì)象時(shí)指定容量防止擴(kuò)容機(jī)制發(fā)生
•  使用原始類(lèi)型集合：相比于包裝類(lèi)內(nèi)存占用減少10倍

// 預(yù)分配容量避免擴(kuò)容
Map<String, Object> dataMap = new HashMap<>(131072); // 2的冪附近
// 使用原始類(lèi)型集合
IntArrayList fastIntList = new IntArrayList(100000);

三.合理利用并發(fā)編程

處理完數(shù)據(jù)存儲(chǔ)上的缺陷后，還是無(wú)法做到同時(shí)處理10w條數(shù)據(jù)，我們可以利用java中的并發(fā)編程讓多個(gè)線(xiàn)程同時(shí)去處理數(shù)據(jù)，這樣，假設(shè)有10個(gè)線(xiàn)程啟動(dòng)，那么每個(gè)線(xiàn)程只需處理1w個(gè)數(shù)據(jù)即可。所以，合理的并發(fā)編程使用極大地幫助我們?nèi)ヌ幚?0w條數(shù)據(jù)。

線(xiàn)程池合理配置

為了能夠最大化利用線(xiàn)程資源，自定義線(xiàn)程池是不二之選，所以線(xiàn)程池的核心參數(shù)配置就尤為重要了，下面我們來(lái)展示下配置的示例。

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    Runtime.getRuntime().availableProcessors(), // 核心線(xiàn)程數(shù)
    Runtime.getRuntime().availableProcessors() * 2, // 最大線(xiàn)程數(shù)
    60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(10000), // 有界隊(duì)列防止內(nèi)存溢出
    new NamedThreadFactory("data-processor"),
    new ThreadPoolExecutor.CallerRunsPolicy() // 飽和策略
);

Runtime.getRuntime().availableProcessors()是配置的關(guān)鍵它的核心價(jià)值在于：

• 動(dòng)態(tài)獲取CPU資源：適應(yīng)不同硬件環(huán)境
• 指導(dǎo)并發(fā)度設(shè)置：線(xiàn)程池、連接池大小
• 實(shí)現(xiàn)彈性伸縮：容器化環(huán)境自動(dòng)適配
• 避免資源浪費(fèi)：防止過(guò)度分配線(xiàn)程

四.分批處理與流式處理

智能分片策略

對(duì)數(shù)據(jù)的處理中，往往涉及與數(shù)據(jù)庫(kù)的操作，如對(duì)訂單的修改或者對(duì)數(shù)據(jù)處理狀態(tài)的記錄等。這時(shí)我們?nèi)绻粭l一條地修改數(shù)據(jù)庫(kù)信息，將會(huì)造成頻繁的數(shù)據(jù)庫(kù)連接造成較大數(shù)據(jù)庫(kù)壓力，此時(shí)，我們可以將多條數(shù)據(jù)分批次進(jìn)行操作。

public class DataShardProcessor {
    public void processInBatches(List<Data> allData, int batchSize) {
        int total = allData.size();
        for (int from = 0; from < total; from += batchSize) {
            int to = Math.min(from + batchSize, total);
            List<Data> batch = allData.subList(from, to);
            // 動(dòng)態(tài)調(diào)整批次大小
            int optimalSize = calculateOptimalBatchSize(batch);
            // 對(duì)數(shù)據(jù)進(jìn)行分批操作
            processBatch(batch, optimalSize);
        }
    }
    private int calculateOptimalBatchSize(List<Data> batch) {
        // 基于數(shù)據(jù)大小、處理復(fù)雜度動(dòng)態(tài)計(jì)算
        return Math.max(100, Math.min(1000, 1000000 / batch.get(0).estimatedSize()));
    }
}

parallelStream并行流處理

平時(shí)我們習(xí)慣使用stream流式處理數(shù)據(jù)，底層是用單線(xiàn)程順序執(zhí)行任務(wù)，當(dāng)遇到map等操作時(shí)就會(huì)去遍歷整個(gè)結(jié)構(gòu)，相當(dāng)耗時(shí)，所以，我們可以使用parallelStream并行流的方式來(lái)提高cpu的利用率，通過(guò)將任務(wù)拆解后合并來(lái)完成任務(wù)。

List<Result> results = dataList.parallelStream()
    .collect(Collectors.groupingByConcurrent(
        Data::getCategory, // 并發(fā)分組
        Collectors.mapping(this::transform, Collectors.toList())
    ))
    .values().parallelStream()
    .flatMap(List::stream)
    .collect(Collectors.toList());

五.消息隊(duì)列解耦

消息隊(duì)列不僅能作為解耦上游接收數(shù)據(jù)和下游處理數(shù)據(jù)的中間層，也是數(shù)據(jù)的緩沖區(qū)以避免下游的系統(tǒng)被沖垮。我們可以通過(guò)部署多個(gè)worker服務(wù)作為消息隊(duì)列的消費(fèi)者，并發(fā)地去隊(duì)列里獲取并處理數(shù)據(jù)。

我們使用RocketMQ來(lái)演示下基本的配置

# RabbitMQ配置優(yōu)化
spring:
  rabbitmq:
    host: localhost
    port: 5672
    # 10萬(wàn)條數(shù)據(jù)需調(diào)優(yōu)
    connection:
      connection-timeout: 10000
    template:
      retry:
        enabled: true
        max-attempts: 3
    listener:
      direct:
        prefetch: 50  # 單次拉取數(shù)量
        concurrency: 4-8  # 消費(fèi)者數(shù)量
        max-concurrency: 8
      simple:
        concurrency: 4-8
        max-concurrency: 8
        retry:
          max-attempts: 3
          stateless: true

六.關(guān)鍵監(jiān)控指標(biāo)和優(yōu)化

關(guān)鍵監(jiān)控指標(biāo)

• 吞吐量指標(biāo)：QPS、TPS、數(shù)據(jù)量/秒
• 延遲指標(biāo)：P50、P90、P99、P999
• 資源指標(biāo)：CPU使用率、GC時(shí)間、堆內(nèi)存、I/O等待
• 業(yè)務(wù)指標(biāo)：處理成功率、錯(cuò)誤類(lèi)型分布

JVM調(diào)優(yōu)關(guān)鍵參數(shù)

# 10萬(wàn)條數(shù)據(jù)處理的JVM建議配置
java -Xms4g -Xmx4g -Xmn2g \  # 固定堆大小避免波動(dòng)
-XX:MaxDirectMemorySize=1g \  # 直接內(nèi)存
-XX:+UseG1GC -XX:MaxGCPauseMillis=200 \  # 低延遲GC
-XX:InitiatingHeapOccupancyPercent=35 \  # 早啟動(dòng)GC
-XX:ParallelGCThreads=4 -XX:ConcGCThreads=2

以上就是Java支撐10W高并發(fā)的架構(gòu)設(shè)計(jì)核心思路的詳細(xì)內(nèi)容，更多關(guān)于Java高并發(fā)架構(gòu)設(shè)計(jì)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論