Apache SkyWalking 中 CounterWindow 并发问题分析与修复

问题背景

在 Apache SkyWalking 的 OAP 服务器中，CounterWindow 类负责处理指标数据的窗口计算。该组件主要用于实现类似 increase('PT1M') 这样的函数调用，用于计算指标在指定时间窗口内的增量值。

问题现象

当多个实例同时上报自定义指标数据时，系统偶尔会抛出 NullPointerException 异常，错误信息显示在比较 Tuple2 对象时出现了空指针。更深入的分析发现，这是由于 PriorityQueue 在多线程环境下被并发修改导致的内部状态损坏。

根本原因分析

CounterWindow 类使用了两个 ConcurrentHashMap 来存储指标数据：

1. lastElementMap 存储最后一个元素
2. windows 存储时间窗口内的数据队列

问题出在 windows 这个 ConcurrentHashMap 上。虽然 ConcurrentHashMap 本身是线程安全的，但它返回的 PriorityQueue 实例并不是线程安全的。当多个线程同时对同一个 PriorityQueue 进行操作时（如 offer、poll、peek 等），会导致队列内部状态不一致，最终引发各种异常。

技术细节

PriorityQueue 是一个基于优先级堆的无界队列，其内部实现依赖于数组和堆排序算法。在多线程环境下，如果不对其进行同步控制，可能会出现以下问题：

1. 数组越界
2. 堆结构破坏
3. 元素丢失
4. 返回错误的最小/最大值

在 SkyWalking 的场景中，当多个线程同时调用 increase() 方法处理同一个指标时，就会对同一个 PriorityQueue 进行并发修改，导致上述问题。

解决方案

最简单的修复方法是在操作 PriorityQueue 时添加同步锁。具体修改是在 increase() 方法中，使用 synchronized 关键字对 window 对象加锁：

public Tuple2<Long, Double> increase(String name, ImmutableMap<String, String> labels, 
                                   Double value, long windowSize, long now) {
    ID id = new ID(name, labels);
    Queue<Tuple2<Long, Double>> window = windows.computeIfAbsent(id, unused -> new PriorityQueue<>());
    synchronized (window) {
        // 所有对window的操作都在同步块内
        window.offer(Tuple.of(now, value));
        // ... 其余逻辑
    }
}

这种解决方案虽然简单，但能有效解决问题。因为：

1. 锁的粒度足够细，只锁定当前操作的队列
2. 不会影响其他指标的并发处理
3. 保持了原有的功能逻辑不变

影响范围

该问题主要影响使用 increase() 函数的自定义指标计算。当系统中有大量实例同时上报相同指标时，问题出现的概率会显著增加。一旦出现，会导致相关指标计算失败，直到服务重启。

最佳实践

对于类似场景的开发，建议：

1. 明确区分容器本身的线程安全性和容器内容的线程安全性
2. 对于复杂的集合操作，考虑是否需要额外的同步措施
3. 在高并发场景下，优先考虑使用专门的并发集合类
4. 进行充分的并发测试，特别是边界条件测试

总结

Apache SkyWalking 中的这个并发问题展示了在多线程环境下使用非线程安全集合的典型风险。通过添加适当的同步控制，可以有效地解决问题，同时保持系统的高性能特性。这也提醒我们在设计类似系统时，需要仔细考虑每个组件的线程安全特性。