1. 框架定位

GroupDataProcessor 是一个通用的 按 Key 分组 → 异步聚合 → 批量处理 框架。核心价值：用可控的延迟换取更高的吞吐量 — 将同一 Key 的高频小数据合并后一次性处理，减少下游调用次数。

框架由四个类构成，职责分明：

2. 四层架构逐层解析

2.1 IGroupData<T> — 数据契约

采用 CRTP 泛型 (T extends IGroupData<T>) 使编译期获得具体子类型，避免向下转型。六个方法构成完整数据生命周期：

2.2 DistributeLocker — 分段锁基础设施

借鉴 ConcurrentHashMap Segment 思想：将无限 key 空间折叠到固定数量的锁槽位上。

key-A 和 key-C 映射到同一 slot 0，逻辑上无关但被串行化 — 这是用可接受的虚假串行化换取固定内存 + 零锁对象GC的设计取舍。

2.3 GroupDataContainer<T> — 线程安全聚合层

持有 ConcurrentHashMap<Object, T> dataMap，通过 DistributeLocker 保证同 key 的 add / copy / clear 操作串行：

2.4 GroupDataProcessor<T> — 调度引擎

将"数据到达"事件转化为"异步处理"任务。核心数据结构：

3. 设计模式与核心思想

3.1 快照隔离 (MVCC 思想)

这是 GDP 最精妙的设计。类比数据库 MVCC：

3.2 递归调度链 (自驱动循环)

GDP 不使用 while(true) 轮询，而是通过回调驱动的递归调度：

3.3 代际标识防重入 (identityObj)

// 每次 scheduleRun 生成唯一标识
Object identityObj = new Object();  // 引用唯一性 = 版本号
taskFutureMap.put(dataKey, new RunningTask(identityObj, future, ...));

// scheduleProcess 完成后校验:
if (runningTask.getIdentityObj() != identityObj) {
    return; // 说明已被超时保护重新调度, 当前任务安静退出
}

这是一个乐观版本号机制。每次 scheduleRun = 一个"代"，identityObj 是该代的唯一身份。用 !=（引用比较）而非 .equals() 是刻意的：new Object() 的引用唯一性本身就是版本号，比 AtomicLong 更轻量且无 ABA 问题。

设计模式汇总

4. 完整数据流

4.1 数据生命周期全景

4.2 并行时序 — 边攒边发

5. 分段锁深度解析

5.1 Hash 算法

private int indexForArray(int length, int h) {
    if ((length & (length - 1)) == 0) {
        // 2 的幂: 扰动函数 + 位掩码 (借鉴 HashMap)
        return (h ^ (h >>> 16)) & (length - 1);
    }
    // 非 2 的幂: 取模
    return Math.abs(h % length);
}

5.2 锁 Metrics 采集

public <T> T lockRun(Object key, FunctionP0<T> runner) {
    Pair<AtomicInteger, Object> lockPair = getLockPair(key);
    AtomicInteger waiting = lockPair.getLeft();
    int count = waiting.incrementAndGet();   // ① 等待计数 +1
    try {
        // ② 上报当前槽位等待线程数
        summary(LOCK_WAITING_DISTRIBUTE, count, tags);
        Timer.Sample sample = Timer.start(); // ③ 开始计时

        synchronized (lockPair.getRight()) { // ④ 获锁 (可能阻塞)
            // ⑤ 上报获锁等待耗时
            time(LOCK_TIME_DISTRIBUTE, sample, tags);
            return runner.run();             // ⑥ 锁内执行
        }
    } finally {
        waiting.decrementAndGet();           // ⑦ 等待计数 -1
    }
}

两个 Metrics 构成锁热度二维视图：

• LOCK_WAITING_DISTRIBUTE (summary): 某槽位此刻排队线程数 → 识别热点锁
• LOCK_TIME_DISTRIBUTE (timer): 从进入到获锁的等待耗时 → 量化锁争用影响

5.3 并发不变式

虽然 scheduleRun 方法本身没有加锁，但它只在两个地方被调用：

1. addData 的 lockRunner 回调中 — 在分段锁内
2. clearData 的 lockRunner 回调中 — 在分段锁内

这是隐式锁传播：scheduleRun 依赖调用者的锁保护。分段锁不仅保护了 dataMap 的数据一致性，也间接保护了 taskFutureMap 中同 key 的调度一致性。

6. 调度机制

6.1 三级延迟策略

long delayTime = clearPushData != null && !clearPushData.isEmpty()
    ? 0                                               // 级别1: 快照未处理完 → 立即
    : (nextDelayTime != null ? nextDelayTime : 1000L); // 级别2: 处理器自定义 / 级别3: 默认1s

6.2 超时保护

6.3 背压与拒绝策略

if (maxSize > 0 && rejectHandler != null && dataContainer.size() >= maxSize) {
    for (T data : dataList) { rejectHandler.run(data); }
    return;
}

7. 关键代码路径解析

7.1 scheduleRun — 延迟投递双路径

private void scheduleRun(Object dataKey, long time, RunningConfig runningConfig) {
    ExecutorService executorService = executorGetter.run(dataKey);
    Object identityObj = new Object();  // 版本号
    Runnable runLogic = () -> scheduleProcess(dataKey, identityObj);
    Future<?> future;
    AtomicReference<Future<?>> secondFutureRef = new AtomicReference<>();

    if (time <= 0) {
        // 路径A: 立即提交
        future = executorService.submit(runLogic);
    } else {
        if (executorService instanceof ScheduledExecutorService) {
            // 路径B: 原生延迟
            future = ((ScheduledExecutorService) executorService)
                     .schedule(runLogic, time, TimeUnit.MILLISECONDS);
        } else {
            // 路径C: 二级中转 — 先定时器延迟, 再提交到业务线程池
            future = this.timeOutScheduler.schedule(() -> {
                Future<?> secondFuture = executorService.submit(runLogic);
                secondFutureRef.set(secondFuture);  // 二级 Future 用于级联 cancel
            }, time, TimeUnit.MILLISECONDS);
        }
    }
    this.taskFutureMap.put(dataKey, new RunningTask(identityObj, future, secondFutureRef));
}

路径C的 secondFutureRef 解决了"定时器 Future 和实际执行 Future 不是同一个"的问题。RunningTask.cancel() 会级联取消两层 Future。

7.2 clearData 回调中的再调度决策

dataContainer.clearData(dataKey, clearPushData, (lockData) -> {
    // lockData: mergeRemove 后的剩余数据, null 表示已清空

    if (lockData == null) {
        // 无剩余数据 → 调度链终止
        taskFutureMap.remove(dataKey);
        return;
    }

    // identityObj 校验: 防止被超时保护替换后的旧任务干扰
    RunningTask runningTask = taskFutureMap.get(dataKey);
    if (runningTask != null && runningTask.getIdentityObj() != identityObj) {
        return; // 已被新一代任务替换, 安静退出
    }

    // 还有数据 → 继续调度
    scheduleRun(dataKey, delayTime, runningConfig);
});

8. GroupDataProcessor vs Disruptor

内存模型对比

9. 优劣与风险评估

9.1 设计优点

9.2 潜在风险

9.3 优化建议

9.4 总评