import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

// 假設(shè)我們有一條交易記錄
class Transaction {
    String id; // 交易唯一ID
    String time; // 交易時間，如 "2023-10-27 09:30:01"
    // ... 其他字段如金額、商戶號等
}

// 定義一個對賬任務(wù)，它繼承自RecursiveTask<對賬結(jié)果>
public class ReconciliationTask extends RecursiveTask<ReconciliationResult> {
    private List<Transaction> bankRecords;
    private List<Transaction> platformRecords;
    private static final int THRESHOLD = 50000; // 閾值，當數(shù)據(jù)量小于5萬時，不再拆分，直接計算

    public ReconciliationTask(List<Transaction> bankRecords, List<Transaction> platformRecords) {
        this.bankRecords = bankRecords;
        this.platformRecords = platformRecords;
    }

    @Override
    protected ReconciliationResult compute() {
        // 如果任務(wù)足夠小，直接執(zhí)行比對
        if (bankRecords.size() <= THRESHOLD && platformRecords.size() <= THRESHOLD) {
            return doDirectReconciliation();
        } else {
            // 否則，進行任務(wù)拆分
            // 示例：按時間范圍簡單地對半拆分（實際生產(chǎn)環(huán)境會更復(fù)雜，比如按小時切分）
            int midBank = bankRecords.size() / 2;
            int midPlatform = platformRecords.size() / 2;

            ReconciliationTask leftTask = new ReconciliationTask(
                bankRecords.subList(0, midBank),
                platformRecords.subList(0, midPlatform)
            );
            ReconciliationTask rightTask = new ReconciliationTask(
                bankRecords.subList(midBank, bankRecords.size()),
                platformRecords.subList(midPlatform, platformRecords.size())
            );

            // 異步執(zhí)行子任務(wù)
            leftTask.fork();
            rightTask.fork();

            // 等待子任務(wù)完成，并合并結(jié)果
            ReconciliationResult leftResult = leftTask.join();
            ReconciliationResult rightResult = rightTask.join();
            return mergeResults(leftResult, rightResult);
        }
    }

    private ReconciliationResult doDirectReconciliation() {
        // 這里是實際的比對邏輯
        // 通常會將一個列表轉(zhuǎn)為Map，以交易ID為Key，實現(xiàn)O(1)的查找
        Map<String, Transaction> platformMap = platformRecords.stream()
            .collect(Collectors.toMap(t -> t.id, t -> t));

        ReconciliationResult result = new ReconciliationResult();

        for (Transaction bankTx : bankRecords) {
            Transaction platformTx = platformMap.get(bankTx.id);
            if (platformTx != null) {
                // 匹配成功，進一步比較金額等細節(jié)
                if (/*金額等細節(jié)一致*/) {
                    result.addMatchedRecord(bankTx, platformTx);
                } else {
                    result.addAmountMismatchRecord(bankTx, platformTx);
                }
                // 從Map中移除，后續(xù)剩下的就是平臺獨有記錄
                platformMap.remove(bankTx.id);
            } else {
                result.addBankOnlyRecord(bankTx);
            }
        }
        // 平臺Map中剩下的記錄都是平臺獨有
        result.addPlatformOnlyRecords(platformMap.values());
        return result;
    }

    private ReconciliationResult mergeResults(ReconciliationResult r1, ReconciliationResult r2) {
        // 合并兩個子任務(wù)的結(jié)果
        // 簡單地將各種類型的記錄列表合并即可
        ReconciliationResult merged = new ReconciliationResult();
        merged.getMatchedRecords().addAll(r1.getMatchedRecords());
        merged.getMatchedRecords().addAll(r2.getMatchedRecords());
        // ... 合并其他如錯配記錄、單邊記錄等
        return merged;
    }
}

// 使用方式
ForkJoinPool forkJoinPool = new ForkJoinPool(); // 默認使用CPU核心數(shù)級別的線程數(shù)
ReconciliationTask mainTask = new ReconciliationTask(allBankRecords, allPlatformRecords);
ReconciliationResult finalResult = forkJoinPool.invoke(mainTask);

val bankRdd: RDD[(String, Transaction)] = sc.parallelize(bankRecords).map(tx => (tx.id, tx))
val platformRdd: RDD[(String, Transaction)] = sc.parallelize(platformRecords).map(tx => (tx.id, tx))

// 內(nèi)連接，得到匹配的記錄
val matchedRdd: RDD[(String, (Transaction, Transaction))] = bankRdd.join(platformRdd)

// 左外連接，然后過濾出平臺為None的，得到銀行單邊記錄
val bankOnlyRdd: RDD[(String, Transaction)] = bankRdd.leftOuterJoin(platformRdd)
                                          .filter { case (id, (bankTx, platformTxOpt)) => platformTxOpt.isEmpty }
                                          .map { case (id, (bankTx, _)) => (id, bankTx) }

// 同理可得平臺單邊記錄
val platformOnlyRdd: RDD[(String, Transaction)] = platformRdd.leftOuterJoin(bankRdd)
                                            .filter { ... }
                                            .map { ... }

// 數(shù)據(jù)預(yù)熱階段：將平臺記錄灌入Redis
Jedis jedis = new Jedis("redis-host");
for (Transaction tx : allPlatformTransactions) {
    // 以交易ID為Key，將交易對象序列化為JSON字符串存儲
    jedis.set(tx.getId(), objectMapper.writeValueAsString(tx));
}

// 對賬階段：遍歷銀行記錄，從Redis中查詢匹配項
for (Transaction bankTx : allBankTransactions) {
    String platformTxJson = jedis.get(bankTx.getId());
    if (platformTxJson != null) {
        Transaction platformTx = objectMapper.readValue(platformTxJson, Transaction.class);
        // 進行細節(jié)比對...
    } else {
        // 銀行單邊賬
    }
}

SELECT b.id, p.id, b.amount, p.amount
FROM bank_transactions b
FULL OUTER JOIN platform_transactions p ON b.id = p.id
WHERE b.amount != p.amount OR b.id IS NULL OR p.id IS NULL;

[對賬文件] -> [數(shù)據(jù)預(yù)處理] -> [按Key（如小時）分片]
                                      |
                                      v
              [內(nèi)存數(shù)據(jù)庫 / 分布式內(nèi)存（如Spark）] <-- 并行計算任務(wù)注入
                                      |
                                      v
        [Task 1: 處理9點數(shù)據(jù)]   [Task 2: 處理10點數(shù)據(jù)]  ... [Task N]
                                      |
                                      v
                              [合并所有任務(wù)結(jié)果]
                                      |
                                      v
                        [生成對賬報告：平、不平、單邊]