采用CAS算法 实现高性能的Disruptor 完成多线程下并发、等待、先后等操作

时间：2022-08-25 13:51:25

来源：https://blog.csdn.net/tianyaleixiaowu/article/details/79787377

拓展：

　　https://www.jianshu.com/p/d24b2eb4a881  初识Disruptor

Java完成多线程间的等待功能：

场景1：一个线程等待其他多个线程都完成后，再进行下一步操作（如裁判员计分功能，需要等待所有运动员都跑完后，才去统计分数。裁判员和每个运动员都是一个线程）。

场景2：多个线程都等待至某个状态后，再同时执行（模拟并发操作，启动100个线程 ，先启动完的需要等待其他未启动的，然后100个全部启动完毕后，再一起做某个操作）。

以上两个场景都较为常见，Java已经为上面的场景1和2分别提供了CountDownLatch和CyclicBarrier两个实现类来完成，参考另一篇文章：https://blog.csdn.net/tianyaleixiaowu/article/details/75234600

而对于更复杂的场景，如

譬如希望1执行完后才执行2，3执行完后才执行4，1和3并行执行，2和4都执行完后才执行last。

还有其他的更奇怪的执行顺序等等。当然这些也可以通过组合多个CountDownLatch或者CyclicBarrier、甚至使用wait、Lock等组合来实现。不可避免的是，都需要使用大量的锁，直接导致性能的急剧下降和多线程死锁等问题发生。那么有没有高性能的无锁的方式来完成这种复杂的需求实现呢？

那就是Disruptor！

Disruptor可以非常简单的完成这种复杂的多线程并发、等待、先后执行等。

至于Disruptor是什么就不说了，直接来看使用：

直接添加依赖包，别的什么都不需要。

       <dependency>

            <groupId>com.lmax</groupId>

            <artifactId>disruptor</artifactId>

            <version>3.4.1</version>

        </dependency>

我只帖关键代码，别的上传到压缩包了。https://download.csdn.net/download/tianyaleixiaowu/10322342

import a.FirstEventHandler;

import a.LongEvent;

import a.LongEventFactory;

import a.LongEventProducer;

import com.lmax.disruptor.BlockingWaitStrategy;

import com.lmax.disruptor.RingBuffer;

import com.lmax.disruptor.dsl.Disruptor;

import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;

import java.util.concurrent.Executors;

import java.util.concurrent.ThreadFactory;

/**

 * @author wuweifeng wrote on 2018/3/29.

 */

public class Simple {

    public static void main(String[] args) {

        ThreadFactory producerFactory = Executors.defaultThreadFactory();

        LongEventFactory eventFactory = new LongEventFactory();

        int bufferSize = 8;

        Disruptor<LongEvent> disruptor = new Disruptor<>(eventFactory, bufferSize, producerFactory,

                ProducerType.SINGLE, new BlockingWaitStrategy());

        FirstEventHandler firstEventHandler = new FirstEventHandler();

        SecondEventHandler secondHandler = new SecondEventHandler();

        ThirdEventHandler thirdEventHandler = new ThirdEventHandler();

        FourthEventHandler fourthEventHandler = new FourthEventHandler();

        LastEventHandler lastEventHandler = new LastEventHandler();

        //1,2,last顺序执行

        //disruptor.handleEventsWith(new LongEventHandler()).handleEventsWith(new SecondEventHandler())

        //        .handleEventsWith(new LastEventHandler());

        //也是1，2，last顺序执行

        //disruptor.handleEventsWith(firstEventHandler);

        //disruptor.after(firstEventHandler).handleEventsWith(secondHandler).then(lastEventHandler);

        //1,2并发执行，之后才是last

        //disruptor.handleEventsWith(firstEventHandler, secondHandler);

        //disruptor.after(firstEventHandler, secondHandler).handleEventsWith(lastEventHandler);

        //1后2，3后4，1和3并发，2和4都结束后last

        disruptor.handleEventsWith(firstEventHandler, thirdEventHandler);

        disruptor.after(firstEventHandler).handleEventsWith(secondHandler);

        disruptor.after(thirdEventHandler).handleEventsWith(fourthEventHandler);

        disruptor.after(secondHandler, fourthEventHandler).handleEventsWith(lastEventHandler);

        disruptor.start();

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

        LongEventProducer longEventProducer = new LongEventProducer(ringBuffer);

        ByteBuffer bb = ByteBuffer.allocate(8);

        for (long i = 0; i < 10L; i++) {

            bb.putLong(0, i);

            longEventProducer.onData(bb);

        }

        disruptor.shutdown();

    }

}

主要就是这个类。我注释掉的部分分别为顺序执行、和12并发然后执行last。

上面那个图对应的代码主要就是after的使用。

运行结果 ：

采用CAS算法 实现高性能的Disruptor 完成多线程下并发、等待、先后等操作

可以看到，由于buffer为8，所以在一个周期内，最大value=7.顺序就是1-3，2-4，1、3是并发的。对于同一个Event，2和4执行完后才执行last。多执行几次看看，就能看明白。

这里用的producer只打印了10个，可以调大，结果就会随机性更好。

相关文章