性能文章>一种极致性能的缓冲队列>

一种极致性能的缓冲队列原创

174101

背景

在多线程下的生产者-消费者模型中,需求满足如下情况:

  • 对生产者生产投递数据的性能要求非常高
  • 多个生产者,单个(多个也可以,本文只介绍单个的情况)消费者
  • 当消费者跟不上生产者速度时,可容忍少部分数据丢失
  • 生产者是单条单条地生产数据

举个日志采集的例子,日志在不同的线程上生产,在日志生产速度远超消费者速度时,可以丢弃部分数据,要求打日志的性能损耗最小,这种情况下可采用本文提供的极致性能的缓冲队列。

实现细节

多个生产者向一个缓冲队列提交消息,说到底是线程安全问题,如果不考虑线程安全,性能必然是最高的,但出现的问题是,数据经常被覆盖。虽然可以容忍少部分数据丢失,但也是在消费者跟不上生产者速度时。缓冲区必然有界,无界可能导致内存泄露,如果缓冲区满,再生产新数据,可选的策略一般有如下几种:

  • 阻塞直到被消费
  • 覆盖旧数据
  • 丢弃新数据

在要求对生产者性能损耗最小的情况下一般不选1,通常采取覆盖策略。

环形队列

有一种环形队列的数据结构(ring buffer)可以很好的解决解决上面提到的生产者-消费者模型、缓冲区有界、覆盖策略。通常用数组来实现ring buffer,只要保证生产者获取下标是线程安全的即可解决线程安全问题。而且数组内存预先分配加上连续内存索引更加快速的特点也保证了强悍的性能。

AtomicInteger

在环形队列上如何保证线程安全地获取数组下标?线程安全地自增我们想到了AtomicInteger,很容易写出如下代码

public class AtomicRangeInteger extends Number {

    private final AtomicInteger value;

    private final int startValue;
    private final int endValue;

    public AtomicRangeInteger(int startValue, int endValue) {
        this.startValue = startValue;
        this.endValue = endValue;
        this.value = new AtomicInteger(startValue);
    }

    public final int incrementAndGet() {
        int next;
        do {
            next = value.incrementAndGet();
            if (next > endValue && value.compareAndSet(next, startValue)) {
                return startValue;
            }
        } while (next > endValue);

        return next;
    }

    public final int get() {
        return value.intValue();
    }

    @Override
    public int intValue() {
        return value.intValue();
    }

    @Override
    public long longValue() {
        return value.intValue();
    }

    @Override
    public float floatValue() {
        return value.intValue();
    }

    @Override
    public double doubleValue() {
        return value.intValue();
    }
}
public final class RingBuffer<T{

    private int bufferSize;
    private AtomicRangeInteger index;
    private final T[] buffer;

    @SuppressWarnings("unchecked")
    public RingBuffer(int bufferSize) {
        this.bufferSize = bufferSize;
        this.index = new AtomicRangeInteger(0, bufferSize);
        this.buffer = (T[]) new Object[bufferSize];
    }

    public final void offer(final T data) {
        buffer[index.incrementAndGet()] = data;
    }

    public final T poll(int index) {
        T tmp = buffer[index];
        buffer[index] = null;
        return tmp;
    }

    public int getBufferSize() {
        return bufferSize;
    }
}

核心代码其实就是这一段

public final int incrementAndGet() {
    int next;
    do {
        next = value.incrementAndGet();
        if (next > endValue && value.compareAndSet(next, startValue)) {
            return startValue;
        }
    } while (next > endValue);

    return next;
}
  • 首选生产者获取下一个可用的index,直接在当前基础上调用incrementAndGet进行加1操作,该操作是原子的,故拿到的一定是没有被其他线程占用的index
  • 获取上一步的返回值,该返回值有可能超过ring buffer的最大下标值,如果超过则将其置为startValue,这一步使用compareAndSet,可能会失败,如果失败说明有其他线程做了该操作,故可以再调用一次incrementAndGet获取下一个下标

为何是极致的性能

目前有一款开源ring buffer的实现——disruptor,关于它的介绍网上可以找到很多,这里简单介绍一下。它也是使用的数组来充当环形队列,但与上面的实现有一点差别,它可以批量插入,所以它使用的是compareAndSet,它在缓冲区填满以后的策略是阻塞,它的下标是不回溯,永远往后加,采用取模来映射到对应的index,为了性能使用位运算(&),所以它的容量只能是2的N次方。

主要的不同就在compareAndSet(v1)与incrementAndGet(v0),我们可以将原先的incrementAndGet实现改为compareAndSet测试一下性能差异(缓冲区大小为1000):

Benchmark                    Mode  Cnt         Score   Error  Units
RingBufferBenchmark.testV0  thrpt    2  39969002.156          ops/s
RingBufferBenchmark.testV1  thrpt    2  15533576.961          ops/s

为什么会有三倍性能的差距?看一下incrementAndGet的实现:

/**
 * Atomically increments by one the current value.
 *
 * @return the updated value
 */

public final int incrementAndGet() {
    return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}
public final int getAndAddInt(Object var1, long var2, int var4) {
    int var5;
    do {
        var5 = this.getIntVolatile(var1, var2);
    } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));

    return var5;
}

发现incrementAndGet底层也是CAS的实现。其实在JDK8之后对incrementAndGet做了优化:

Unsafe是经过特殊处理的,不能理解成常规的java代码,在调用getAndAddInt的时候,如果系统底层支持fetch-and-add,那么它执行的就是native方法,使用的是fetch-and-add;如果不支持,就按照上面的所看到的getAndAddInt方法体那样,以java代码的方式去执行,使用的是compare-and-swap

CAS使用的是jdk层面的自旋锁,fetch-and-add是cpu指令,性能上fetch-and-add要强很多。所以如果拿上面的代码用jdk1.7来测试,会发现性能没有差别。

再优化空间

  • 缓存行填充

AtomicRangeInteger对象中存在三个属性,value,startValue,endValue,value是经常变化的,startValue,endValue是不会变化,所以当value经常变化会导致读取startValue,endValue时不会命中cpu缓存,性能有所下降,我们使用jdk8的注解@Contended来填充value所在行的缓存。

public class AtomicRangeIntegerV2 extends Number {

    @Contended
    protected final AtomicInteger value;

    protected final int startValue;
    protected final int endValue;
    ...
}

做一下benchmark(v2为加@Contended注解):

Benchmark                    Mode  Cnt         Score   Error  Units
RingBufferBenchmark.testV2  thrpt    2  72095754.040          ops/s
RingBufferBenchmark.testV0  thrpt    2  44360926.943          ops/s
  • 多个ring buffer分担冲突

这个优化就不做过多说明了,多线程中一般都采取分段的思想来降低冲突,ring buffer也可以,当一个ring buffer存在性能瓶颈时,可以利用多个ring buffer来分担,最佳状态是每个线程分配一个ring buffer,具体怎么分配,在之前的文章《实现一个比LongAdder更高性能的计数器有多难?》中有一个比较巧妙的办法,可以参考下。

彩蛋

这个ring buffer本来是借鉴skywalking中ring buffer的实现,但当时skywalking的实现也是使用CAS,性能不是很满意,于是我就做了这些优化,具体github的issue可以参考如下链接:

https://github.com/apache/skywalking/pull/2874 https://github.com/apache/skywalking/pull/2930

现在skywalking中已经是我写的这个版本了,算得上是一个极致性能的缓冲队列了。

 

请先登录,感受更多精彩内容
快去登录吧,你将获得
  • 浏览更多精彩评论
  • 和开发者讨论交流,共同进步

为你推荐

字符串字面量长度是有限制的
前言 偶然在一次单元测试中写了一个非常长的字符串字面量。 正文 在一次单元测试中,我写了一个很长的字符串字面量,大概10万个字符左右,编译时,编译器给出了异常告警 `java: constant
多次字符串相加一定要用StringBuilder而不用-吗?
今天在写一个读取Java class File并进行分析的Demo时,偶然发现了下面这个场景(基于oracle jdk 1.8.0_144): ``` package test; public c
如何通过反射获得方法的真实参数名(以及扩展研究)
前段时间,在做一个小的工程时,遇到了需要通过反射获得方法真实参数名的场景,在这里我遇到了一些小小的问题,后来在部门老大的指导下,我解决了这个问题。通过解决这个问题,附带着我了解到了很多新的知识,我觉得
高吞吐、低延迟 Java 应用的 GC 优化实践
本篇原文作者是 LinkedIn 的 Swapnil Ghike,这篇文章讲述了 LinkedIn 的 Feed 产品的 GC 优化过程,虽然文章写作于 April 8, 2014,但其中的很多内容和
「每日五分钟,玩转 JVM」:久识你名,初居我心
聊聊 JVMJVM,一个熟悉又陌生的名词,从认识Java的第一天起,我们就会听到这个名字,在参加工作的前一两年,面试的时候还会经常被问到JDK,JRE,JVM这三者的区别。JVM可以说和我们是老朋友了
据说99.99%的人都会答错的类加载的问题
概述首先还是把问题抛给大家,这个问题也是我厂同学在做一个性能分析产品的时候碰到的一个问题。 同一个类加载器对象是否可以加载同一个类文件多次并且得到多个Class对象而都可以被java层使用吗请仔细注意
Java多线程——并发测试
编写并发程序时候,可以采取和串行程序相同的编程方式。唯一的难点在于,并发程序存在不确定性,这种不确定性会令程序出错的地方远比串行程序多,出现的方式也没有固定规则。那么如何在测试中,尽可能的暴露出这些问
Java多线程知识小抄集(一)
本文主要整理笔者遇到的Java多线程的相关知识点,适合速记,故命名为“小抄集”。本文没有特别重点,每一项针对一个多线程知识做一个概要性总结,也有一些会带一点例子,习题方便理解和记忆。 1.interr