并发编程 | 并发编程框架 - Disruptor - 深入理解高性能异步处理框架

原创

kfaino

发布于 2023-11-04 09:35:34

7070

发布于 2023-11-04 09:35:34

文章被收录于专栏：程序员的日常程序员的日常

总览

本章节的思维导图如下所示：

前言

在并发编程的世界中，对效率的追求从未停止过。我们尝试用各种方式来提高程序的执行效率，包括使用更高级的并发控制结构，如锁和线程池，以及采用更先进的并发设计模式。然而，有一种工具在许多高性能系统中得到了广泛的应用，那就是Disruptor。Disruptor是一个高性能的异步处理框架，它利用了Ring Buffer、CAS等高效的并发策略，使得在处理高并发、低延迟的需求时，表现出了惊人的性能。

在这篇博客中，我们将一起深入探索Disruptor的内部工作原理，分析其如何提供出类似于常见队列但是性能卓越的数据结构，以及为何能在高并发环境下实现高效的数据交换。我们也将通过实例展示如何在实际项目中使用Disruptor，以帮助我们更好地理解其使用方法和性能优势。让我们一起开启这个高性能异步处理框架的探索之旅吧！

我们来看一张图：

这是一张展示了Disruptor和ArrayBlockingQueue性能对比的直方图，你可以明显地看到Disruptor的优越性能。如果你对它感兴趣，可以跳转到这个链接查看what_is_the_disruptor

2010年的3月，在伦敦QCon上，LMAX提到Disruptor框架。次年，Martin Fowler在它的博客 The LMAX Architecture 中探讨有关LMAX架构。主要讲述的是LMAX，这个新的零售金融交易平台，它每天需要处理大量的交易。为了应对这个大麻烦，LMAX团队研发了一款Disruptor框架。

这个堪称神器的框架居然是2010年甚至之前的产物，我们赶紧了解下....

入门 | Disruptor高性能的秘密

Disruptor，我对它的定义为“并发破局者”，是 LMAX 公司开发的一种高性能，低延迟的并发框架。它采用了一些独特的设计理念，比如零阻塞，预分配数据，确保数据的局部性，使得在一些高吞吐量，低延迟的场景中，能够表现出优秀的性能。

上面有几个关键的设计理念，我们来分析一下：

零阻塞

在并发编程中，线程阻塞通常是由等待资源（例如锁或数据）引起的。想象一下你在餐馆用餐，如果服务员在为其他顾客服务，你可能需要等待，这种等待就是阻塞。在 Disruptor 中，设计者采用了无锁的设计，这就好比餐馆有足够的服务员为每个顾客服务，所以你不需要等待，可以直接享用你的美食，这就是"零阻塞"。

预分配数据

在并发编程中，动态数据分配可能会成为一个性能瓶颈，因为为对象分配内存和初始化可能需要消耗一定的时间。而 Disruptor 通过将数据预先分配在 RingBuffer 中，使得每个处理线程在处理数据时，数据在多个 CPU 之间的传递被降到了最低，从而提高了性能。预分配数据就像在一场音乐会开始前，你已经预先为所有的观众分配了座位。这样，当他们到来时，他们可以直接坐下，无需等待工作人员找位置。这就提高了系统的处理速度。

确保数据的局部性

在现代的计算机硬件中，为了提高处理速度，CPU 会将常用的数据存储在一个叫做缓存的地方。如果数据在多个 CPU 之间频繁地传递，那么 CPU 就需要不断地更新缓存，这就造成了所谓的“缓存行击穿”，会消耗很多时间。在 Disruptor 中，数据尽可能地在同一个 CPU 中处理，就像我们在工作时，我们希望所有需要的文件和资料都在我们的办公桌上，这样我们就不需要来回跑去取文件，从而提高了工作效率。

入门 | 使用Disruptor框架

接下来，我们来看下如何使用Disruptor，我在下文为你贴出代码，你可以看一下。

首先，我们定义了一个事件，以及事件对应的工厂：

public class LongEvent {
    private long value;

    public long getValue() {
        return value;
    }

    public void setValue(long value) {
        this.value = value;
    }
}

public class LongEventFactory implements EventFactory<LongEvent> {
    public LongEvent newInstance() {
        return new LongEvent();
    }
}

紧接着，生产者和消费者出现了。

public class LongEventProducer {
    private final RingBuffer<LongEvent> ringBuffer;

    public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    public void onData(long value) {
        long sequence = ringBuffer.next();
        try {
            LongEvent event = ringBuffer.get(sequence);
            event.setValue(value);
        } finally {
            ringBuffer.publish(sequence);
        }
    }
}

public class LongEventConsumerHandler implements EventHandler<LongEvent> {
    public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {
        System.out.println("事件值: " + event.getValue());
    }
}

我们来测试看看：

public class DisruptorTest {
    public static void main(String[] args) throws Exception {
        Executor executor = Executors.newCachedThreadPool();
        LongEventFactory factory = new LongEventFactory();
        int bufferSize = 1024;

        Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, executor);
        disruptor.handleEventsWith(new LongEventHandler());
        disruptor.start();

        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        LongEventProducer producer = new LongEventProducer(ringBuffer);

        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long l = 0; l<100; l++) {
            bb.putLong(0, l);
            producer.onData(bb.getLong(0));
            // 延迟一秒
            Thread.sleep(1000);
        }
    }
}

结果如下：

Connected to the target VM, address: '127.0.0.1:2144', transport: 'socket'
事件值: 0
事件值: 1
事件值: 2
事件值: 3
...
Disconnected from the target VM, address: '127.0.0.1:2144', transport: 'socket'

事件值有条不紊的输出。当然，为了方便你理解，我把生产者和消费者都为你注明。若你在工作中要想使用它，建议你把这两个角色从需求中抽象出来，好的设计往往事半功倍。

虽然你已经熟悉了Disruptor框架的使用，但仍有一层微妙的薄雾笼罩在你对它的全面理解上。现在，让我们一起揭开这层薄雾，深入探索其底层原理。

进阶 | Disruptor 的工作流程

在此之前，我建议你对上面的代码有一定的印象。当然，我也会为你把代码贴出来，你可以结合着理解。现在，我们开始。

为了更好的理解，我画了一张图，你可以看一下：

初始化

首先，我们需要初始化一个 Ring Buffer 和一组 EventProcessor（事件处理器）。代码如下：

Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, executor);

Ring Buffer 是一个环形的数据结构，可以存储并传输数据（事件）。EventProcessor 是处理事件的消费者。

生产事件

生产者通过 Ring Buffer 的 next() 方法获取一个事件槽（slot），假如是图中的slot-2。这个方法返回的是一个序列号，表示分配给这个事件的唯一标识。代码如下：

long sequence = ringBuffer.next();

填充事件

生产者使用这个序列号获取对应的事件槽，并在其中填充数据。这个步骤可以是生产者自行完成，也可以是通过 Ring Buffer 的 publishEvent() 方法进行。我们上面的示例是通过onData()方法完成。

我们结合另一张图，继续往下看：

发布事件

当事件数据填充完成后，生产者需要调用 Ring Buffer 的 publish() 方法来发布这个事件。这个方法会更新 Ring Buffer 中的序列号，表示新的事件已经准备好并可供消费。代码如下：

finally {
          ringBuffer.publish(sequence);
      }

消费事件

每个 EventProcessor 都有一个 SequenceBarrier（序列屏障）。这个序列屏障会持续监控 Ring Buffer 的序列号。消费者会在序列屏障的 waitFor() 方法处阻塞，直到有新的事件发布。

处理事件

当 Ring Buffer 的序列号更新（即有新的事件发布）后，waitFor() 方法会返回最新的可用序列号，消费者就可以开始处理对应的事件。

更新消费者序列

消费者处理完事件后，需要更新其自身的序列，表示它已经完成了对应的事件处理。这通常是通过 EventProcessor 的 onEvent() 方法来完成的。代码如下：

    public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {
        System.out.println("事件值: " + event.getValue());
    }

重复过程

上述的过程会不断重复，生产者和消费者会持续在 Ring Buffer 中进行事件的生产和消费。

多生产者和多消费者

单生产者单消费者只是为了方便你理解而举得特例，实际我们在使用的时候，它可以支持多生产者和多消费者的模型。生产者和消费者各自有一个序列号来跟踪它们在 RingBuffer 中的位置。生产者间通过 CAS（Compare And Swap）操作安全地发布事件，消费者通过序列屏障来确定何时可以安全地读取事件。

高级 | Disruptor 源码分析

还是老规矩，以问题出发来看源码。上面文章讲到Disruptor高性能的秘密在于它独特的设计理念。我们来回顾一下，有哪些设计理念：零阻塞，预分配数据，数据的局部性

接下来，我们一个一个进行分析。

零阻塞

上面的代码中，我们是通过next()方法来获取sequence。理所应当的，我们也来分析这个方法。

    public long next()
    {
        return sequencer.next();
    }

很简单的代码，通过Sequencer来调用next()方法。因为是单生产者，所以分支走到SingleProducerSequencer我们接着往下看：

public long next(int n)
{
    // 检查参数n是否大于等于1，n表示我们想要生产的事件的数量
    if (n < 1)
    {
        throw new IllegalArgumentException("n must be > 0");
    }

    // 获取下一个序列号
    long nextValue = this.nextValue;

    // 计算下一个事件的序列号和超过缓冲区大小的位置
    long nextSequence = nextValue + n;
    long wrapPoint = nextSequence - bufferSize;
    // 获取缓存的gatingSequence
    long cachedGatingSequence = this.cachedValue;

    // 如果wrapPoint大于缓存的gatingSequence，或者缓存的gatingSequence大于当前的序列号，那么就会发生自旋等待
    if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue)
    {
        cursor.setVolatile(nextValue);  // StoreLoad fence

        long minSequence;
        // 自旋等待，直到wrapPoint不再大于minSequence
        while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences, nextValue)))
        {
            LockSupport.parkNanos(1L); // 使用LockSupport.parkNanos(1L)实现自旋等待
        }

        // 更新缓存的gatingSequence
        this.cachedValue = minSequence;
    }

    // 更新下一个序列号
    this.nextValue = nextSequence;

    // 返回下一个事件的序列号
    return nextSequence;
}

这段代码的主要设计目标是确保生产者不会生产出消费者还未消费的事件，以此来保证零阻塞。它通过维护一个cachedValue，并使用自旋等待来实现这一目标。如果消费者已经消费了足够多的事件，那么生产者就可以生产更多的事件。如果消费者还没有消费足够的事件，那么生产者就会进入自旋等待，直到消费者消费了足够的事件。

接着，我们来看下消费端的waitFor()方法：

public long waitFor(final long sequence)
    throws AlertException, InterruptedException, TimeoutException
{
    // 检查是否有警报，如果有，抛出AlertException
    checkAlert();

    // 通过waitStrategy.waitFor方法实现等待策略，返回当前可用的最大序列号
    long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);

    // 如果当前可用的最大序列号小于消费者希望消费的序列号，那么消费者只能消费到当前可用的最大序列号
    if (availableSequence < sequence)
    {
        return availableSequence;
    }

    // 如果当前可用的最大序列号大于等于消费者希望消费的序列号，消费者可以尝试消费更多的事件，方法是通过获取当前已经发布的最大序列号
    return sequencer.getHighestPublishedSequence(sequence, availableSequence);
}

这段代码的主要设计目标是确保消费者只消费生产者已经生产出的事件。如果生产者还没有生产出消费者希望消费的事件，那么消费者需要等待。等待的具体策略由waitStrategy.waitFor实现，这也是实现"零阻塞"设计的关键。如果你对它感兴趣，可以阅读源码，查看每个策略的实现。篇幅有限，我这里就不展开讨论。

依然是篇幅有限原因，后面的预分配数据，数据的局部性 我就不讨论了，就当给各位的作业，如果您对它感兴趣，欢迎你去阅读源码。

至此，源码已经为你分析完了。现在你是不是已经掌握了Disruptor 框架了呢？

进阶 | Disruptor Vs ArrayBlockingQueue 性能分析

还记得最开始的那张图吗？我为你展示Disruptor和·ArrayBlockingQueue两者的性能差异。但是，光图可不够，我们得自己验证一下。老规矩，上代码。

首先是Disruptor 测试，我们直接来模拟亿级的“set”操作。测试代码如下：

        Executor executor = Executors.newCachedThreadPool();
        LongEventFactory factory = new LongEventFactory();
        int bufferSize = 1024;

        Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, executor);
        disruptor.handleEventsWith(new LongEventConsumerHandler());
        disruptor.start();

        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        LongEventProducer producer = new LongEventProducer(ringBuffer);

        long start = System.currentTimeMillis();
        for (long l = 0; l < 100000000; l++) {
            producer.onData(l);
        }

        long end = System.currentTimeMillis();
        System.out.println("Disruptor 花费时间: " + (end - start) + " ms");

执行结果：

Connected to the target VM, address: '127.0.0.1:2874', transport: 'socket'
Disruptor 花费时间: 31507 ms

再看看下ArrayBlockingQueue：

ArrayBlockingQueue<Long> queue = new ArrayBlockingQueue<>(1024);

        // 消费者线程
        new Thread(() -> {
            while (true) {
                try {
                    Long l = queue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

        long start = System.currentTimeMillis();
        for (long l = 0; l < 10000000; l++) {
            queue.put(l);
        }
        long end = System.currentTimeMillis();

        System.out.println("ArrayBlockingQueue 花费时间: " + (end - start) + " ms");

结果如下：

Connected to the target VM, address: '127.0.0.1:2650', transport: 'socket'
ArrayBlockingQueue 花费时间: 38728 ms

这....好像差不不太大啊，那我还不如用我熟悉的ArrayBlockingQueue 降低学习门槛。别急，我们接着往下看。

查阅官网后，和大量博客文献进行分析，我们得知：Disruptor 的优势在于能够处理大量的并发事件和对高吞吐率的支持。接下来，我们从这两个关键的点来讨论：

任务性质

如果任务本身需要消耗大量的 CPU 时间，那么并发框架的选择可能不会对总体性能产生太大影响。相反，如果任务主要是 I/O 绑定，或者包含大量的等待，那么 Disruptor 由于其设计的非阻塞性，可能会显著提高性能。

并发水平

当并发的线程数量增加时，一些传统的并发容器可能会因为线程上下文切换的成本和锁竞争而性能下降。在这种情况下，Disruptor 的无锁设计和对线程调度的优化可能会带来显著的性能提升。

原来是这样啊，看来不能单纯的看时间的多少，还得看场景。

以并发水平为例，我们来进行测试。代码如下：

ExecutorService executorService = Executors.newCachedThreadPool();
        LongEventFactory factory = new LongEventFactory();
        int bufferSize = 1024;

        Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, executorService);
        disruptor.handleEventsWith(new LongEventConsumerHandler(), new LongEventConsumerHandler());
        disruptor.start();

        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        List<LongEventProducer> longEventProducers = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            longEventProducers.add(new LongEventProducer(ringBuffer));
        }

        long start = System.currentTimeMillis();

        CountDownLatch latch = new CountDownLatch(10);

        for (int i = 0; i < 10; i++) {
            int finalI = i;
            executorService.execute(() -> {
                for (long l = 0; l < 10000000; l++) {
                    longEventProducers.get(finalI).onData(l);
                }
                latch.countDown();
            });
        }

        latch.await();

        long end = System.currentTimeMillis();
        System.out.println("Disruptor 花费时间: " + (end - start) + " ms");

我们开辟了10个生产者来进行“set”值。老规矩，看下结果。

Connected to the target VM, address: '127.0.0.1:8515', transport: 'socket'
Disruptor 花费时间: 33714 ms

和上面单线程的差别不大，如果你在ArrayBlockingQueue中使用多线程，你会看到一个惊人的数字：

Connected to the target VM, address: '127.0.0.1:9340', transport: 'socket'
ArrayBlockingQueue 花费时间: 256174 ms

性能分析到这里就结束了。读到这里，你应该越发觉得框架在使用场景上的选型到底有多重要了吧。接下来，我们来看下使用场景和注意事项。

进阶 | Disruptor 的适用场景和使用注意事项

看完之后，你是不是想把Disruptor 放到生产去实践？别急，看完这些你再考虑也不迟。

Disruptor的适用场景

高并发、低延迟

Disruptor最初是为高性能、低延迟的场景设计的。它在一些金融交易、游戏、实时计算等需要高并发、低延迟的场景中有广泛的应用。

多生产者、多消费者

Disruptor支持多生产者、多消费者模型，适合需要大量并发读写的场景。

无锁设计

在需要大量并发操作，但又希望避免锁带来的性能损耗的场景中，可以考虑使用Disruptor。

事件驱动的架构

Disruptor是基于事件驱动的编程模型设计的，如果你的应用架构是事件驱动的，Disruptor是一个不错的选择。它能有效地处理和分发大量的事件，确保事件的快速和准确处理。

使用Disruptor的注意事项

谨慎配置

Disruptor的性能和正确性很大程度上依赖于配置。例如，buffer的大小、等待策略等都需要根据实际场景进行合理配置。

正确处理异常

在Disruptor中，任何一个消费者抛出的异常都可能会影响到其它消费者。因此，需要在消费者中正确处理异常，避免出现一处异常导致整个系统出问题的情况。

注意内存管理

Disruptor的高性能部分来自于它对内存的有效管理。但这也要求开发者在使用Disruptor时要更加注意内存管理，避免出现内存泄漏等问题。

明确使用场景

虽然Disruptor在某些场景下表现优异，但并不是所有场景都适合使用Disruptor。在某些情况下，使用Java自带的并发工具可能更加简单有效。

对缓存行填充有一定了解

因为Disruptor的设计利用了缓存行填充（false sharing）来提高性能，所以最好对此有一定的理解。如果读者对伪共享感兴趣，后续我会专门出一篇文章进行讲解。

总结

好，我们来回顾一下。首先，我向你揭示了Disruptor框架高性能的秘密。然后，我们一起深入探讨了Disruptor框架本身。在此基础上，我们深入剖析了Disruptor的工作原理，这让我们对源码的理解更加深入。最后，我们分析了Disruptor和ArrayBlockingQueue在性能上的差异，并对在这个过程中需要注意的问题进行了详细的说明。如果你计划在生产环境中使用Disruptor，我希望你能充分考虑这些问题。至此，本篇结束。

附录：相关资源和进一步阅读

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

并发编程

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

并发编程

登录后参与评论

0 条评论

热度