欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

互联网技术16——Disruptor

程序员文章站 2022-07-11 12:22:40
...

学之前

在看这篇博客之前,我想说的是,如果是准备入门Disruptor,建议掌握一些重要方法和特性,至于要实现哪种功能,建议掌握大致流程,以后的使用过程中慢慢去消化,如果学习任务比较多,又期望通过自己的入门式学习而全部掌握Disruptor并熟练运用时间成本有些划不来的。所以我准备有时间做一个简单归纳,这篇博客介绍了几种使用案例,同时推荐一个Disruptor的学习网站

http://ifeve.com/disruptor-info/

 

开始学习:

Disruptor是一个java的BlockQueue,它的目的就是在相同处理器的不同线程之间move数据(message or event)。相对于一般的queue,disruptor具有一些重要的特性,为事件分配内存,可选的lock-free、根据消费依赖路径多路发送事件给消费者。

Disruptor说明文档:https://github.com/LMAX-Exchange/disruptor/wiki/Introduction

互联网技术16——Disruptor

LMAX是一种新型零售金融交易平台,它能够以很低的延迟产生大量交易。这个系统是建立在JVM平台上,其核心是一个业务逻辑处理器,它能够在一个线程里每秒处理600万订单。业务逻辑处理器完全是运行在内存中,使用事件源驱动的方式。业务逻辑驱动器的核心是Disruptor。Disruptor主动将数据发送到数据消费端,600万指的是分发给消费端的订单数。

Disruptor是一个开源的并发框架,并获得2011年的程序框架大赛创新奖,能够在无锁的情况下实现网络queue并发操作,disruptor是一个高性能的异步处理矿建,或者认为是最快的消息框架(轻量的JMS),也可以是一个观察者模式的实现,或者时间监听模式的实现。

相关术语:
 

RingBuffer:它是Disruptor最主要的组件,从3.0开始RingBuffer仅仅负责存储和更新在Disruptor中流通的数据,对一些特殊的使用场景能够被用户(使用其他数据结构)完全替代。

Sequence:表示一个特殊组件处理的序号,每个消费者(EventProcessor)都维持着一个Sequence,大部分的并发代码依赖这些Sequence值运转,因此Sequence支持多种当前为AtomicLong类的特性。

Sequencer:这是Disruptor真正的核心,实现了这个接口的两种生产者(单生产者和多生产者)均实现了所有的并发算法,为了在生产者和消费者之间进行准确快速的数据传递。

SequenceBarrier:由Sequencer生成,并且包含了已经发布的Sequence的引用,它包含了决定是否有供消费者来消费的Event的逻辑。

WaitStrategy:决定一个消费者将如何等待生产者将Event置入Disruptor。

Event:从生产者到消费者过程中所处理的数据单元,它完全由用户自己定义。

EventHandler:由用户实现并且代表了Disruptor中的一个消费者的接口。

Producer:由用户实现,它调用RingBuffer来插入事件,在Disruptor中没有其相应的实现代码,由用户实现。

WorkProcessor:确保每个sequence只被一个processor消费,在同一个WorkPool中处理多个WorkerProcessor不会消费同样的sequence。

WorkerPool:一个WorkProcessor池,其中WorkProcessor将消费sequence,所以任务可以在实现了WorkHandler接口的worker之间移交。

 

RingBuffer是一个首尾相接的环,可以把它作为在不同上下文(线程)之间传递数据的buffer。RingBuffer中拥有一个序号Sequence,它指向下一个可用元素。随着Producer不停地填充数据到这个buffer(可能也会有读取),这个序号会一直增长,直到绕过这个环。要找到数组中当前序号指向的元素,可以通过mod操作实现,即sequence mod array length = array index 。RingBuffer槽的个数最好为2的n次方,这样有利于基于二进制的计算机进行计算。RingBuffer使用的是数组的顺序存储结构,所以比链表更快。

 

在Disruptor中,创建一个HelloWorld示例程序的步骤:

1.建立一个Event类

2.建立一个工厂Event类,用于创建Event类实例对象

3.建立一个监听事件类,用于消费处理Event类型的数据

4.main函数测试代码编写,实例化Disruptor实例,配置一系列参数。然后,对Disruptor实例绑定监听事件类,接收并处理数据

5.在Disruptor中,真正存储数据的核心叫做RingBuffer,通过Disruptor实例获取RingBuffer,生产者Producer将数据生产出来,将数据加入到RingBuffer的槽中

LongEvent.java:

package com.disruptorTest;

/**
 * Created by BaiTianShi on 2018/8/25.
 */
public class LongEvent {
    //需要通过事件传递的业务数据
    private long value;

    public long getValue() {
        return value;
    }

    public void setValue(long value) {
        this.value = value;
    }
}
LongEventFactory.java
package com.disruptorTest;

import com.lmax.disruptor.EventFactory;

/**
 * Created by BaiTianShi on 2018/8/25.
 */

//我们需要给disruptor传递一个时间工厂,让disruptor就可以为我们创建事件
public class LongEventFactory implements EventFactory {
    @Override
    public Object newInstance() {
        return new LongEvent();
    }
}
LongEventProducer.java
package com.disruptorTest;


import com.lmax.disruptor.RingBuffer;

import java.nio.ByteBuffer;

/**
 * Created by BaiTianShi on 2018/8/25.
 *
 * 发布事件最少需要两步,1:获取一个事件槽;2:发布事件
 * 发布事件的时候要使用try/finally来保证事件一定被发布
 * 如果我们使用ringBuffer获取一个时间槽,那么一定要发布对应的事件
 * 如果不发布事件,那么就会引起低收入disruptor的状态混乱
 * 尤其是在多个事件生产者的情况下,会导致事件消费者失速,从而不得不重启应用才能恢复
 */

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


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

    /**
     * onData   用来发布事件,每调用一次就发布一次事件
     * 它的参数用事件传递给消费者
     */
    public void onData(ByteBuffer buffer){
        //1.可以把ringBuffer看做一个事件队列,那么next就是得到下面一个事件槽
        long  sequence = ringBuffer.next();

        try {
            //2.用上面的索引取出一个空事件用于填充(获取该序号对应的事件对象)
            LongEvent event = ringBuffer.get(sequence);
            //3.获取要通过对象传递的业务数据
            event.setValue(buffer.getLong(0));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //4.发布事件
            //注意,最后的ringBuffer.publisth一定要包含在finally中,以确保必须得到调用
            //某个请求的sequence未被提交,将会阻塞后续的发布操作或者其他的producer
            ringBuffer.publish(sequence);
        }
    }
}

 

LongEventHandler.java
package com.disruptorTest;


import com.lmax.disruptor.EventHandler;

/**
 * Created by BaiTianShi on 2018/8/25.
 */
//消费者,事件处理器。这里只是简单的将数据打印到控制台
public class LongEventHandler implements EventHandler<LongEvent> {
    @Override
    public void onEvent(LongEvent longEvent, long l, boolean b) throws Exception {
        System.out.println(longEvent.getValue());
    }
}

main.java

package com.disruptorTest;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by BaiTianShi on 2018/8/25.
 */
public class LongEventMain {

    public static void main(String[] args) {
        //创建线程池
        ExecutorService executorService = Executors.newCachedThreadPool();
        //创建longEvent工厂
        LongEventFactory factory = new LongEventFactory();

        //设置ringBuffer大小,必须为2的n指数被,这样才能通过位运算,保证速度
        int ringBufferSize = 1024*1024;

        //最后一个参数为指定声明disruptor的策略
        //BlockingWaitStrategy  最低效的策略,但其对cpu4de消耗最小,并且在各种环境下能提供更加一致的性能表现
        //SleepingWaitStrategy 性能和BlocklingWaitStrategy性能差不多,对cpu消耗也雷士,但其对生产者现场影响最小,适合用于异步日志类似的场景
        //YieldingWaitStrategy 性能是最好的,适合用于递延次的系统中,在要求极高性能且事件处理线数小于cpu逻辑核心数的场景中,推荐使用。例如:cpu开启超线程的特性

        //创建disruptor, longEvent为指定的消费数据类型,即每个事件槽中存放的数据类型
        Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(factory,ringBufferSize,executorService,ProducerType.SINGLE, new YieldingWaitStrategy());

        //连接消费事件
        disruptor.handleEventsWith(new LongEventHandler());
        //启动
        disruptor.start();

        //获取具体的存放数据的ringBuffer
        RingBuffer ringBuffer = disruptor.getRingBuffer();

        LongEventProducer producer = new LongEventProducer(ringBuffer);
//        LongEventProducerWithTranslator producer  = new LongEventProducerWithTranslator(ringBuffer);
        ByteBuffer buf = java.nio.ByteBuffer.allocate(8);
        for(int i=0; i<10; i++){
            buf.putLong(0,i);
            producer.onData(buf);
        }
        disruptor.shutdown();//关闭disruptor,方法会阻塞,直到所有事件都得到处理
        executorService.shutdown();//关闭线程池,如果需要的话必须手动关闭,disruptor在shutdown时不会自动关闭线程池
    }
}

运行结果:

0
1
2
3
4
5
6
7
8
9

互联网技术16——Disruptor

 

将main中的代码LongeventProducer = new LongEventPeoducer(ringBuffer)修改为

LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);

同时将LongeventProducer修改为LongEventProducerWithTranslator

package com.disruptorTest;

import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;

import java.nio.ByteBuffer;

/**
 * Created by BaiTianShi on 2018/8/27.
 *
 * disruptor 3.0提供了lambda式api,这样可以把一些复杂的操作放在RingBuffer中
 * 所以在disruptor3.0之后的版本最好使用event publisher或者event Translator来发布事件
 */
public class LongEventProducerWithTranslator {

    private RingBuffer ringBuffer;
    //一个translator可以看做一个事件初始化器,publicEvent方法会调用它
    //填充Event
    private  static final EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR =
            new EventTranslatorOneArg<LongEvent, ByteBuffer>() {
                @Override
                public void translateTo(LongEvent longEvent, long l, ByteBuffer buffer) {
                    longEvent.setValue(buffer.getLong(0));
                }
            };

    public LongEventProducerWithTranslator(RingBuffer ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    void onData(ByteBuffer buffer){
        ringBuffer.publishEvent(TRANSLATOR,buffer);
    }
}

 

直接使用RingBuffer(使用消息处理器)

trade.java

package com.ringBufferTest;

import java.math.BigDecimal;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by BaiTianShi on 2018/8/27.
 */
public class Trade {
    private String id;

    private String name ;

    private BigDecimal price;

    private AtomicInteger count = new AtomicInteger(0);

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public BigDecimal getPrice() {
        return price;
    }

    public void setPrice(BigDecimal price) {
        this.price = price;
    }
}

TradeHandler.java  事件处理器

package com.ringBufferTest;

import com.lmax.disruptor.EventHandler;

import java.util.UUID;

/**
 * Created by BaiTianShi on 2018/8/27.
 */
public class TradeHandler implements EventHandler<Trade> {
    @Override
    public void onEvent(Trade trade, long l, boolean b) throws Exception {
        System.out.println("执行TradeHandler中onEvent(Trade event)方法的线程名"+Thread.currentThread().getName());
        //这里做具体的消费逻辑
        trade.setId(UUID.randomUUID().toString());//简单生成下ID
        System.out.println(trade.getId());

    }
}

main

package com.ringBufferTest;

import com.lmax.disruptor.*;

import java.math.BigDecimal;
import java.util.concurrent.*;

/**
 * Created by BaiTianShi on 2018/8/27.
 */
public class Main1 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        int BUFFER_SIZE = 1024;
        int THREAT_NUMBERS=4;

        //createSingleProducer 创建一个单生产者的ringBuffer
        //第一个参数叫EvenFactory,从名字上理解就是“事件工厂”,其实它的职责就是产生数据填充ringBuffer的区块,
        //第二个参数是ringBuffer的大小,必须是2的指数倍,目的是为了求模运算为&运算提高效率
        //第三个参数是RingBuffer的生产者在没有区块的时候(可能是消费者(或者说是事件处理器))太慢了的等待策略
        final RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<Trade>() {
            @Override
            public Trade newInstance() {
                return new Trade();
            }
        },BUFFER_SIZE,new YieldingWaitStrategy());

        //创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(THREAT_NUMBERS);

        //创建sequenceBarrier
        SequenceBarrier sb = ringBuffer.newBarrier();

        //创建消息处理器
        BatchEventProcessor<Trade> tradeBatchEventProcessor = new BatchEventProcessor<>(ringBuffer,sb,new TradeHandler());

        //这一步的目的是把消费者的位置引用注入到生产者中,如果只有一个消费者的情况下可以省略
        ringBuffer.addGatingSequences(tradeBatchEventProcessor.getSequence());

        executorService.submit(tradeBatchEventProcessor);

        System.out.println("主线程名"+Thread.currentThread().getName());
        //如果存在多个消费者 那重复执行上面3行代码 把TradeHandler换成其它消费者类

        Future<?> future = executorService.submit(new Callable<Void>() {
            @Override
            public Void call() throws Exception {
                long seq;
                System.out.println("执行call()方法的线程名"+Thread.currentThread().getName());
                for(int i=0;i<10;i++){
                    seq = ringBuffer.next();
                    Trade trade =ringBuffer.get(seq);
                    trade.setPrice(new BigDecimal(i));
                    ringBuffer.publish(seq);
                }
                return null;
            }
        });

        future.get();//等待生产者结束
        Thread.sleep(10000);//等上10秒,等消费都处理完成
        tradeBatchEventProcessor.halt();//通知事件(或者说消息)处理器 可以结束了(并不是马上结束!!!)
        executorService.shutdown();//终止线程  
    }

}

运行结果

互联网技术16——Disruptor

 

直接使用RingBuffer(使用WorkHandler与WorkerPool)

Trade.java

package com.ringBufferTest;

import java.math.BigDecimal;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by BaiTianShi on 2018/8/27.
 */
public class Trade {
    private String id;

    private String name ;

    private BigDecimal price;

    private AtomicInteger count = new AtomicInteger(0);

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public BigDecimal getPrice() {
        return price;
    }

    public void setPrice(BigDecimal price) {
        this.price = price;
    }
}

TradeHandler2.java

package com.ringBufferTest;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;

import java.util.UUID;

/**
 * Created by BaiTianShi on 2018/8/27.
 */
public class TradeHandler2 implements WorkHandler<Trade> {
    @Override
    public void onEvent(Trade trade) throws Exception {
        System.out.println("执行TradeHandler中onEvent(Trade event)方法的线程名"+Thread.currentThread().getName());
        //这里做具体的消费逻辑
        trade.setId(UUID.randomUUID().toString());//简单生成下ID
        System.out.println(trade.getId());

    }
}

main

package com.ringBufferTest;

import com.lmax.disruptor.*;

import java.math.BigDecimal;
import java.util.concurrent.*;

/**
 * Created by BaiTianShi on 2018/8/27.
 */
public class Main2 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        int BUFFER_SIZE = 1024;
        int THREAT_NUMBERS=4;

        //createSingleProducer 创建一个单生产者的ringBuffer
        //第一个参数叫EvenFactory,从名字上理解就是“事件工厂”,其实它的职责就是产生数据填充ringBuffer的区块,
        //第二个参数是ringBuffer的大小,必须是2的指数倍,目的是为了求模运算为&运算提高效率
        //第三个参数是RingBuffer的生产者在没有区块的时候(可能是消费者(或者说是事件处理器))太慢了的等待策略
        final RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<Trade>() {
            @Override
            public Trade newInstance() {
                return new Trade();
            }
        },BUFFER_SIZE,new YieldingWaitStrategy());

        //创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(THREAT_NUMBERS);
        SequenceBarrier sq = ringBuffer.newBarrier();
        WorkHandler<Trade> workHandler = new TradeHandler2();

        WorkerPool pool = new WorkerPool(ringBuffer,sq,new IgnoreExceptionHandler(),workHandler);
        pool.start(executorService);

        System.out.println("主线程名"+Thread.currentThread().getName());
        //如果存在多个消费者 那重复执行上面3行代码 把TradeHandler换成其它消费者类

       for(int i = 0; i< 8; i++){
           long seq = ringBuffer.next();
           Trade t = ringBuffer.get(seq);
           t.setPrice(new BigDecimal(i));
           ringBuffer.publish(seq);
       }

        Thread.sleep(10000);//等上10秒,等消费都处理完成
        pool.halt();//通知事件(或者说消息)处理器 可以结束了(并不是马上结束!!!)
        executorService.shutdown();//终止线程
    }

}

运行结果:

互联网技术16——Disruptor

使用Disruptor实现菱形、六边形操作

Trade.java

package com.DisruptorLiuBianXing;

import java.math.BigDecimal;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by BaiTianShi on 2018/8/27.
 */
public class Trade {
    private String id;

    private String name ;

    private BigDecimal price;

    private AtomicInteger count = new AtomicInteger(0);

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public BigDecimal getPrice() {
        return price;
    }

    public void setPrice(BigDecimal price) {
        this.price = price;
    }
}

TradePublisher.java

package com.DisruptorLiuBianXing;

import com.lmax.disruptor.dsl.Disruptor;

import java.util.concurrent.CountDownLatch;

/**
 * Created by BaiTianShi on 2018/8/28.
 */
public class TradePublisher implements Runnable {
    private Disruptor<Trade> disruptor;
    private CountDownLatch downLatch;

    private static int LOOP = 1;

    public TradePublisher(Disruptor<Trade> disruptor, CountDownLatch downLatch) {
        this.disruptor = disruptor;
        this.downLatch = downLatch;
    }

    @Override
    public void run() {
        TradeEventTranslator tradeEventTranslator = new TradeEventTranslator();
        for(int i=0; i<LOOP; i++){
            disruptor.publishEvent(tradeEventTranslator);
        }
        downLatch.countDown();
    }
}

TradeEventTranslator.java

package com.DisruptorLiuBianXing;

import com.lmax.disruptor.EventTranslator;

import java.math.BigDecimal;
import java.util.Random;

/**
 * Created by BaiTianShi on 2018/8/28.
 */
public class TradeEventTranslator implements EventTranslator<Trade> {
    private Random random = new Random();


    @Override
    public void translateTo(Trade trade, long l) {
        this.generateTrade(trade);
    }

    private  Trade generateTrade(Trade trade){
        trade.setPrice(new BigDecimal(random.nextDouble()).multiply(new BigDecimal(9999)));
        return trade;
    }
}

Handler1.java

package com.DisruptorLiuBianXing;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;

/**
 * Created by BaiTianShi on 2018/8/28.
 */
public class Handler1 implements EventHandler<Trade>,WorkHandler<Trade> {
    @Override
    public void onEvent(Trade trade, long l, boolean b) throws Exception {
        onEvent(trade);
    }

    @Override
    public void onEvent(Trade trade) throws Exception {
        System.out.println("handle1:set name");
        trade.setName("h1");
        Thread.sleep(1000);
    }
}

Handler2.java

package com.DisruptorLiuBianXing;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;

import java.math.BigDecimal;

/**
 * Created by BaiTianShi on 2018/8/28.
 */
public class Handler2 implements EventHandler<Trade>,WorkHandler<Trade> {
    @Override
    public void onEvent(Trade trade, long l, boolean b) throws Exception {
        onEvent(trade);
    }

    @Override
    public void onEvent(Trade trade) throws Exception {
        System.out.println("handle2:set price");
        trade.setPrice(new BigDecimal(17));
        Thread.sleep(1000);
    }
}

Handler3.java

package com.DisruptorLiuBianXing;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;

/**
 * Created by BaiTianShi on 2018/8/28.
 */
public class Handler3 implements EventHandler<Trade>,WorkHandler<Trade> {
    @Override
    public void onEvent(Trade trade, long l, boolean b) throws Exception {
        onEvent(trade);
    }

    @Override
    public void onEvent(Trade trade) throws Exception {
        System.out.println("handler3: name: " + trade.getName() + " , price: " + trade.getPrice() + ";  instance: " + trade.toString());
    }
}

Main.java

package com.DisruptorLiuBianXing;

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by BaiTianShi on 2018/8/28.
 */
public class Main {
    public static void main(String[] args) throws InterruptedException {
        long beginTime = System.currentTimeMillis();
        int bufferSize = 1024;

        ExecutorService executor = Executors.newFixedThreadPool(8);

        Disruptor<Trade> disruptor = new Disruptor<Trade>(new EventFactory<Trade>() {
            @Override
            public Trade newInstance() {
                return new Trade();
            }
        },bufferSize,executor, ProducerType.SINGLE,new BusySpinWaitStrategy());

        //菱形操作

        //使用disruptor创建消费者组C1,C2
        EventHandlerGroup<Trade> handlerGroup = disruptor.handleEventsWith(new Handler1(), new Handler2());
        handlerGroup.then(new Handler3());


        //顺序操作
        /**
         disruptor.handleEventsWith(new Handler1()).
         handleEventsWith(new Handler2()).
         handleEventsWith(new Handler3());
         */

        //六边形操作.
        /**
         Handler1 h1 = new Handler1();
         Handler2 h2 = new Handler2();
         Handler3 h3 = new Handler3();
         Handler4 h4 = new Handler4();
         Handler5 h5 = new Handler5();
         disruptor.handleEventsWith(h1, h2);
         disruptor.after(h1).handleEventsWith(h4);
         disruptor.after(h2).handleEventsWith(h5);
         disruptor.after(h4, h5).handleEventsWith(h3);
         */

        //启动
        disruptor.start();

        CountDownLatch latch = new CountDownLatch(1);
        executor.submit(new TradePublisher(disruptor,latch));

        latch.await();

        disruptor.shutdown();
        executor.shutdown();;
        System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));
    }
}

运行结果:

互联网技术16——Disruptor

 

生产者消费者模式:

Order为Disruptor中的RingBuffer环中每一个槽存放数据的类型。Producer用于产生事件并放置到RingBuffer事件槽中。Consumer是用于消费RingBuffer中事件的类。

Order.java

package com.disruptorOrder;

/**
 * Created by BaiTianShi on 2018/8/28.
 */
public class Order {
    private String id;

    private String name;

    private double price;

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public double getPrice() {
        return price;
    }

    public void setPrice(double price) {
        this.price = price;
    }
}

producer.java

package com.disruptorOrder;

import com.lmax.disruptor.RingBuffer;

/**
 * Created by BaiTianShi on 2018/8/28.
 */
public class Producer {
    private final RingBuffer<Order> ringBuffer;

    public Producer(RingBuffer<Order> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    /**
     * onData用来发布事件,每调用一次就发布一次事件
     * 它的参数会用过事件传递给消费者
     */
    public void onData(String data){
        //可以把ringBuffer看做一个事件队列,那么next就是得到下面一个事件槽
        long sequence = ringBuffer.next();

        try {
            //用上面的索引取出一个空的事件用于填充(获取该序号对应的事件对象)
            Order order = ringBuffer.get(sequence);
            //获取要通过事件传递的业务数据
            order.setId(data);
        } finally {
            //发布事件
            //注意,最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用;如果某个请求的 sequence 未被提交,将会堵塞后续的发布操作或者其它的 producer
            ringBuffer.publish(sequence);
        }

    }
}

Consumer.java

package com.disruptorOrder;

import com.lmax.disruptor.WorkHandler;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by BaiTianShi on 2018/8/28.
 */
public class Consumer implements WorkHandler<Order> {
    private String consumerId;
    private static AtomicInteger count = new AtomicInteger(0);

    public Consumer(String consumerId) {
        this.consumerId = consumerId;
    }

    @Override
    public void onEvent(Order order) throws Exception {
        System.out.println("当前消费者"+consumerId+"消费消息"+order.getId());
        count.incrementAndGet();
    }
    public int getCount(){
        return count.get();
    }

}

main.java

package com.disruptorOrder;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by BaiTianShi on 2018/8/28.
 */
public class Main {
    //提供Integer的原子操作,在高并发场景中,多线程操作能够保证线程安全
    private static AtomicInteger count = new AtomicInteger(0);

    public static void main(String[] args) throws InterruptedException {
        //创建ringBuffer
        RingBuffer<Order> buffer = RingBuffer.create(ProducerType.MULTI, new EventFactory<Order>() {
            @Override
            public Order newInstance() {
                return new Order();
            }
        }, 1024 * 1024, new YieldingWaitStrategy());

        //生产者消费者协调操作
        SequenceBarrier barrier = buffer.newBarrier();

        //消费者数组
        Consumer[] consumers = new Consumer[3];
        for (int i = 0; i < consumers.length; i++) {
            consumers[i] = new Consumer("c"+i);
        }

        WorkerPool<Order> workerPool = new WorkerPool<Order>(buffer,barrier, new IgnoreExceptionHandler(),consumers);

        //这一步的目的是把消费者的位置信息引用注入的生产者中,协调生产者和消费者的速度
        buffer.addGatingSequences(workerPool.getWorkerSequences());

        workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));

        final CountDownLatch latch = new CountDownLatch(1);
        for (int i = 0; i < 3; i++) {
           final Producer p = new Producer(buffer);

            new Thread(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+"ready");
                    try {
                        latch.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    for (int j = 0; j < 2; j++) {
                        p.onData(Thread.currentThread().getName()+("产生的事件数"+String.valueOf(count.addAndGet(1))));
                    }
                }
            }).start();
        }

        Thread.sleep(2000);
        System.out.println("开始生产");
        //终结上面两次循环中的等待
        latch.countDown();
        Thread.sleep(5000);
        System.out.println("总数"+consumers[0].getCount());
    }
}

运行结果:

互联网技术16——Disruptor

 

相关标签: 多线程