Disruptor基本使用步骤

Disruptor是一个高性能的异步处理框架，一个轻量级的JMS，和JDK中的BlockingQueue有相似处，但是它的处理速度非常快，获得2011年程序框架创新大奖，号称“一个线程一秒钟可以处理600W个订单”，并且Disruptor不仅仅只有buffer，它提供的功能非常强大。

disruptor开发步骤:

1. 定义event队列中需要处理的元素

2. 定义event工厂,用于填充处理队列

3. 定义消费者

引入依赖:

        
            com.lmax
            disruptor
            3.3.2
        

定义event队列中需要处理的元素

package com.cyc.mystudy.disruptor;


public class StringEvent {

    private String stringEvent;


    public void setStringEvent(String stringEvent) {
        this.stringEvent = stringEvent;
    }

    public String getStringEvent() {
        return stringEvent;
    }
}

定义event工厂,用于填充处理队列

package com.cyc.mystudy.disruptor;

import com.lmax.disruptor.EventFactory;


public class StringEventFactory implements EventFactory {
    @Override
    public StringEvent newInstance() {
        return new StringEvent();
    }
}

定义消费者

package com.cyc.mystudy.disruptor;

import com.lmax.disruptor.EventHandler;


public class StringEventHander implements EventHandler {
    @Override
    public void onEvent(StringEvent stringEvent, long l, boolean b) throws Exception {
        System.out.println(Thread.currentThread().getName()+":"+stringEvent.getStringEvent());
    }
}

创建TestDisruptor类

public static void main(String[] args) {
    //生产者
    StringEventFactory factory=new StringEventFactory();
    //大小
    int bufferSize=1024;
    //定义
    Disruptor disruptor = new Disruptor<>(factory, bufferSize,  Executors.newCachedThreadPool());
    //连接消费端方法
    disruptor.handleEventsWith(new StringEventHander());
    //启动
    disruptor.start();
    //拿到实际存储队列
    RingBuffer ringBuffer = disruptor.getRingBuffer();

for (int i=0;i<1024;i++){
    long next = ringBuffer.next();
    StringEvent stringEvent = ringBuffer.get(next);
    stringEvent.setStringEvent(String.valueOf(i));
    //提交发布操作
    ringBuffer.publish(next);
}
}

disruptor的几种等待策略:

 想要使用等待策略上面创建disruptor可以使用如下方法创建

Disruptor disruptor =
new Disruptor(factory, bufferSize, Executors.newCachedThreadPool(), ProducerType.SINGLE, new YieldingWaitStrategy());

如果有多个消费者怎么指定:

StringEventHander stringEventHander = new StringEventHander();
StringEventHander stringEventHander1 = new StringEventHander();
//连接多个消费端方法
disruptor.handleEventsWith(stringEventHander,stringEventHander1);

多个消费者由多个线程执行

如果消费者遇到异常出报错:

我们在消费者中抛出一个异常

public class StringEventHander implements EventHandler {
    @Override
    public void onEvent(StringEvent stringEvent, long l, boolean b) throws Exception {
        System.out.println(Thread.currentThread().getName()+":"+stringEvent.getStringEvent());
        throw new Exception("测试异常");
    }
}

此时运行则会:

怎么处理:

我们在TestDisruptor加入如下代码: 

StringEventHander stringEventHander = new StringEventHander();
//连接多个消费端方法
disruptor.handleEventsWith(stringEventHander);
disruptor.handleExceptionsFor(stringEventHander).with(new ExceptionHandler() {
    @Override
    public void handleEventException(Throwable throwable, long l, StringEvent stringEvent) {
        System.out.println(throwable.getMessage());
    }
    @Override
    public void handleOnStartException(Throwable throwable) {
    }
    @Override
    public void handleOnShutdownException(Throwable throwable) {
    }
});

这里处理异常的方式是打印异常的信息

 此时运行:

就不会出错 

这里重写3个方法分别对应

 运行时出现异常 

启动时异常

 关闭时异常的处理方案