codahale 指标 Meter mark() 方法如何线程安全？

Question

我最近开始学习 CodaHale/DropWizard 指标库。我不明白 Meter 类的线程安全性如何（根据文档），尤其是 mark() 和 tickIfNecessary() 方法：

https://github.com/dropwizard/metrics/blob/3.2-development/metrics-core/src/main/java/com/codahale/metrics/Meter.java#L54-L77

public void mark(long n) {
    tickIfNecessary();
    count.add(n);
    m1Rate.update(n);
    m5Rate.update(n);
    m15Rate.update(n);
}

private void tickIfNecessary() {
    final long oldTick = lastTick.get();
    final long newTick = clock.getTick();
    final long age = newTick - oldTick;
    if (age > TICK_INTERVAL) {
        final long newIntervalStartTick = newTick - age % TICK_INTERVAL;
        if (lastTick.compareAndSet(oldTick, newIntervalStartTick)) {
            final long requiredTicks = age / TICK_INTERVAL;
            for (long i = 0; i < requiredTicks; i++) {
                m1Rate.tick();
                m5Rate.tick();
                m15Rate.tick();
            }
        }
    }
}

我可以看到有一个 AtomicLong 类型的 lastTick，但仍然可能存在 m1-m15 速率的滴答声稍长一些的情况，因此另一个线程可以调用这些滴答声以及下一个 TICK_INTERVAL 的一部分。由于 Rates 的 tick() 方法根本不同步，这难道不是一种竞争条件吗？ https://github.com/dropwizard/metrics/blob/3.2-development/metrics-core/src/main/java/com/codahale/metrics/EWMA.java#L86-L95

public void tick() {
    final long count = uncounted.sumThenReset();
    final double instantRate = count / interval;
    if (initialized) {
        rate += (alpha * (instantRate - rate));
    } else {
        rate = instantRate;
        initialized = true;
    }
}

谢谢，

玛丽安

Answer 1

这是线程安全的，因为来自tickIfNecessary() 的这一行每个newIntervalStartTick 只返回一次true

if (lastTick.compareAndSet(oldTick, newIntervalStartTick))

如果两个线程几乎同时进入tickIfNecessary()会发生什么？

两个线程从oldTick 读取相同的值，确定至少TICK_INTERVAL 纳秒已经过去并计算newIntervalStartTick。

现在两个线程都尝试做lastTick.compareAndSet(oldTick, newIntervalStartTick)。正如名称compareAndSet 所暗示的那样，此方法将lastTick 的当前值与oldTick 进行比较，并且仅当该值等于oldTick 时，它才会自动替换为newIntervalStartTick 并返回true。

由于这是一条原子指令（在硬件级别！），因此只有一个线程可以成功。当另一个线程执行此方法时，它已经将newIntervalStartTick 视为lastTick 的当前值。由于这个值不再匹配oldTick，更新失败并且方法返回false，因此这个线程不会调用m1Rate.tick()到m15Rate.tick()。

EWMA.update(n) 方法使用java.util.concurrent.atomic.LongAdder 来累积事件计数，从而提供类似的线程安全保证。

Answer 2

据我所知，你是对的。如果调用tickIfNecessary() 使得age > TICK_INTERVAL 在另一个调用仍在运行时，则有可能从多个线程同时调用m1Rate.tick() 和其他tick() 方法。所以归结为tick() 及其调用的例程/操作是安全的。

让我们剖析tick()：

public void tick() {
    final long count = uncounted.sumThenReset();
    final double instantRate = count / interval;
    if (initialized) {
        rate += (alpha * (instantRate - rate));
    } else {
        rate = instantRate;
        initialized = true;
    }
}

alpha 和 interval 仅在实例初始化时设置，并标记为 final 那些线程安全的，因为它们是只读的。 count 和 instantRate 是本地的，无论如何其他线程都不可见。 rate 和 initialized 被标记为 volatile 并且这些写入应该始终对后续读取可见。

如果我没记错的话，从initialized 的第一次读取到initialized 或rate 的最后一次写入几乎都可以进行比赛，但有些没有效果，例如当 2 个线程为将initialized 切换为true。

似乎大多数有效的比赛都可以发生在rate += (alpha * (instantRate - rate)); 中，尤其是丢弃或混合计算，例如：

1. 假设：initialized 是 true
2. Thread1：计算count，instantRate，检查initialized，第一次读取rate，我们称之为previous_rate，无论出于何种原因，都停止了
3. Thread2：计算count，instantRate，检查initialized，并计算rate += (alpha * (instantRate - rate));
4. Thread1：继续其操作并计算rate += (alpha * (instantRate - previous_rate));

如果以某种方式对读取和写入进行排序，以便在所有线程上读取rate，然后在所有线程上写入，则会发生丢弃，从而有效地丢弃一个或多个计算。

但是这种竞争的概率，这意味着 age > TICK_INTERVAL 匹配，使得 2 个线程运行到相同的 tick() 方法，尤其是 rate += (alpha * (instantRate - rate)) 可能非常低，并且取决于不明显的值。

只要LongAdderProxy 对update/add 和sumThenReset 中的tick() 方法使用线程安全的数据结构，mark() 方法似乎就是线程安全的。

我认为唯一能够回答未解决的问题的人——无论比赛没有明显影响还是以其他方式减轻——是项目作者或对项目的这些部分和计算的值有深入了解的人。