标签：dsl produce state sleep 覆盖 throws cer def 多个
Disruptor是一款java高性能无锁并发处理框架。和JDK中的BlockingQueue有相似处，但是它的处理速度非常快！！！号称“一个线程一秒钟可以处理600W个订单”（反正渣渣电脑是没体会到）。
Disruptor功能十分强大，比如消费者阻塞等待；生产者-消费者一对一、一对多、多对一、多对多；构建消费者串/并行处理链等等。
具体的概念模型可以参考：https://www.cnblogs.com/haiq/p/4112689.html
下面是我基于Disruptor框架封装的工具。采用fluent编码风格，简化了Disruptor的调用。
package com.gravel.demo.test.disruptor;
import com.gravel.demo.test.disruptor.base.EventProducer;
import com.gravel.demo.test.disruptor.base.Publisher;
import com.gravel.demo.test.disruptor.base.PublisherBuilder;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description: Disruptor 高性能异步处理框架
*/
public class DisruptorTest {
public static void main(String[] args) throws Exception {
builderTest();
}
private static void builderTest() throws Exception {
// 创建多个生产者实例
EventProducer<Domain> producer1 = new EventProducer<>("producer1");
EventProducer<Domain> producer2 = new EventProducer<>("producer2");
// 创建多个消费者实例
DomainConsumer handler1 = new DomainConsumer("handler1");
DomainConsumer handler2 = new DomainConsumer("handler2");
DomainConsumer after1 = new DomainConsumer("after1");
DomainConsumer after2 = new DomainConsumer("after2");
DomainConsumer after3 = new DomainConsumer("after3");
DomainConsumer then = new DomainConsumer("then");
// 创建消息发布者
final Publisher<Domain> publisher = PublisherBuilder.newBuilder()
// 设置线程工厂
// .threadFactory(r -> new Thread(r))
// .threadFactory(new LimitedThreadFactory())
// .threadFactory(Executors.defaultThreadFactory())
// 设置生产类型
// .producerType(ProducerType.SINGLE)
// .producerType(ProducerType.MULTI)
// 设置事件工厂
// .eventFactory(new EventFactory())
// 设置等待策略
// .waitStrategy(new SleepingWaitStrategy())
// .waitStrategy(new YieldingWaitStrategy())
// 设置发布方式
// .publishStrategy(PublishStrategy.TRANSLATOR)
// .publishStrategy(PublishStrategy.NORMAL)
// 设置ringBuffer大小
// .ringSize(1024 * 8)
// 设置异常处理器
.exceptionHandler(new DomainErrorHandler<>())
// 初始化Disruptor, 在配置生产者和消费者之前一定要先初始化。
.disruptor()
// 设置单生产者
// .producer(producer1)
// 配置单个消费者
// .handler(handler1)
// 配置多生产者
.producer(producer1, producer2)
// ====== 设置多个workers或者handlers处理链 start =======
// .worker(handler1)
.handler(handler1, handler2)
.after(handler1).handler(after1)
.after(handler2).handler(after2)
.after(after1, after2).handler(after3)
// .then(after3)
// ====== 设置多个workers或者handlers处理链 end =======
// 启动
.build();
long start = System.currentTimeMillis();
try {
for (int i = 0; i < 500; i++) {
publisher
// 可连续发布
//.publish(new Domain(String.valueOf("a" + i), "init"))
.publish(new Domain(String.valueOf(i), "init"));
}
} finally {
long sleep = 200;
Thread.sleep(sleep);
System.out.println("used time: " + (System.currentTimeMillis() - start - sleep) + "ms");
// 不关闭会一直阻塞等待
publisher.shutdown();
}
}
}
从上面的代码来看，我们封装的工具类入口是Publisher。他可以配置一系列Disruptor需要的参数，如线程工厂（ThreadFactory）、事件工厂（EventFactory）、等待策略（WaitStrategy）、消息生产者（Producer）、消费者（Handler/Worker）等等。
其中消息生产者和消费者是Publisher的关键，所以稍后重点描述。先看看其他PublisherBuilder类。
package com.gravel.demo.test.disruptor.base;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description:
*/
public class PublisherBuilder {
// Publish 默认参数
private static final int RING_SIZE = 1024 * 8;
private static final ThreadFactory THREAD_FACTORY = Executors.defaultThreadFactory();
private static final WaitStrategy WAIT_STRATEGY = new SleepingWaitStrategy();
private static final com.lmax.disruptor.EventFactory EVENT_FACTORY = new EventFactory();
private static final ProducerType PRODUCER_TYPE = ProducerType.SINGLE;
private static final PublishStrategy PUBLISH_STRATEGY = PublishStrategy.TRANSLATOR;
private com.lmax.disruptor.EventFactory eventFactory;
private ThreadFactory threadFactory;
private WaitStrategy waitStrategy;
private ExceptionHandler exceptionHandler;
private ProducerType type;
private PublishStrategy publishStrategy;
private EventPublisher publisher;
private int ringSize;
public static PublisherBuilder newBuilder() {
return new PublisherBuilder();
}
/**
* 指定ringBuffer size,最好为2的n次方。默认1024*8
*
* @param ringSize
* @return
*/
public PublisherBuilder ringSize(int ringSize) {
this.ringSize = ringSize;
return this;
}
/**
* 指定eventFactory， 默认EventFactory
*
* @param eventFactory
* @param <T>
* @return
*/
public <T> PublisherBuilder eventFactory(com.lmax.disruptor.EventFactory eventFactory) {
this.eventFactory = eventFactory;
return this;
}
/**
* 指定ThreadFactory, 默认Executors.defaultThreadFactory()
*
* @param threadFactory
* @return
*/
public PublisherBuilder threadFactory(ThreadFactory threadFactory) {
this.threadFactory = threadFactory;
return this;
}
/**
* 指定等待策略, 默认SleepingWaitStrategy
*
* @param waitStrategy
* @return
*/
public PublisherBuilder waitStrategy(WaitStrategy waitStrategy) {
this.waitStrategy = waitStrategy;
return this;
}
public PublisherBuilder publishStrategy(PublishStrategy publishStrategy) {
this.publishStrategy = publishStrategy;
return this;
}
/**
* 初始化disruptor
*
* @return
*/
public PublisherBuilder disruptor() {
this.eventFactory = this.eventFactory == null ? EVENT_FACTORY : this.eventFactory;
this.threadFactory = this.threadFactory == null ? THREAD_FACTORY : this.threadFactory;
this.waitStrategy = this.waitStrategy == null ? WAIT_STRATEGY : this.waitStrategy;
this.ringSize = this.ringSize <= 0 ? RING_SIZE : this.ringSize;
this.type = this.type == null ? PRODUCER_TYPE : this.type;
this.publishStrategy = this.publishStrategy == null ? PUBLISH_STRATEGY : this.publishStrategy;
publisher = new EventPublisher<>(eventFactory, ringSize, threadFactory, waitStrategy, exceptionHandler, type, publishStrategy);
return this;
}
/**
* 配置生产者
* @param producers
* @param <T>
* @return
*/
public <T> PublisherBuilder producer(EventProducer<T> ...producers) {
if (isInit()) {
this.publisher.producer(producers);
}
return this;
}
/**
* eventHandler：每个event事件可以被所有handler处理
*
* @param eventHandlers
* @param <T>
* @return
*/
public <T> PublisherBuilder handler(EventHandler<Event<T>>... eventHandlers) {
if (isInit()) {
this.publisher.eventHandler(eventHandlers);
}
return this;
}
/**
* workHandler：每个event事件只能被一个work处理
*
* @param workHandlers
* @param <T>
* @return
*/
public <T> PublisherBuilder worker(WorkHandler<Event<T>>... workHandlers) {
if (isInit()) {
this.publisher.workHandler(workHandlers);
}
return this;
}
/**
* handler或work之后处理
*
* @param thenEventHandlers
* @param <T>
* @return
*/
public <T> PublisherBuilder then(EventHandler<Event<T>>... thenEventHandlers) {
if (isInit()) {
this.publisher.thenHandler(thenEventHandlers);
}
return this;
}
/**
* 顺序指定handler，不可接直接在worker方法后面调用
*
* @param afterEventHandlers
* @param <T>
* @return
*/
public <T> PublisherBuilder after(EventHandler<Event<T>>... afterEventHandlers) {
if (isInit()) {
this.publisher.afterHandler(afterEventHandlers);
}
return this;
}
/**
* 指定producerType
*
* @param type
* @param <T>
* @return
*/
public <T> PublisherBuilder producerType(ProducerType type) {
this.type = type;
return this;
}
/**
* 异常处理类
*
* @param exceptionHandler
* @param <T>
* @return
*/
public <T> PublisherBuilder exceptionHandler(ExceptionHandler<T> exceptionHandler) {
this.exceptionHandler = exceptionHandler;
return this;
}
private boolean isInit() {
if (this.publisher == null) {
throw new IllegalStateException("execute disruptor() function before set handlers or workers.");
}
return true;
}
public <T> Publisher<T> build() {
return this.publisher.start();
}
}
其中关键代码是disruptor()方法，即Disruptor实例化入口。因为eventHandler和workHandler的指定必须在disruptor实例之后，disruptor.start()启动之前。
所以我们调用工具类必须如下：
PublisherBuilder.newBuilder()/*.各种配置*/.disruptor()/*指定producer和handler/worker*/.build();
来看看PublisherBuilder中的EventPulisher类， 他实现了Publisher接口， 如下：
Publisher.java
package com.gravel.demo.test.disruptor.base;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description:
*/
public interface Publisher<T> {
Publisher<T> start();
Publisher<T> publish(T t);
Publisher<T> shutdown();
}
EventPublisher.java
package com.gravel.demo.test.disruptor.base;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;
import java.util.Objects;
import java.util.concurrent.ThreadFactory;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description:
*/
public class EventPublisher<T> implements Publisher<T> {
private RingBuffer<Event<T>> ringBuffer;
private Disruptor<Event<T>> disruptor;
private EventHandlerGroup<Event<T>> handlerGroup;
private PublisherState state;
private PublishStrategy publishStrategy;
private EventTranslator<T> translator;
private Producer<T>[] producers;
private enum PublisherState {
START, SHUTDOWN
}
public EventPublisher(com.lmax.disruptor.EventFactory<Event<T>> factory,
int ringSize,
ThreadFactory threadFactory,
WaitStrategy waitStrategy,
ExceptionHandler<Event<T>> exceptionHandler,
ProducerType type,
PublishStrategy publishStrategy) {
this.disruptor = new Disruptor<>(factory, ringSize, threadFactory,
type, waitStrategy);
if (!Objects.isNull(exceptionHandler)) {
this.disruptor.setDefaultExceptionHandler(exceptionHandler);
}
this.ringBuffer = disruptor.getRingBuffer();
this.publishStrategy = publishStrategy;
if (publishStrategy == PublishStrategy.TRANSLATOR) {
translator = new EventTranslator<>();
}
this.state = PublisherState.SHUTDOWN;
}
public EventPublisher<T> producer(EventProducer<T> ...producers) {
if (!Objects.isNull(producers) && producers.length > 0) {
for (EventProducer<T> producer : producers) {
producer.setRingBuffer(this.ringBuffer).setTranslator(this.translator);
}
this.producers = producers;
}
return this;
}
public EventPublisher<T> eventHandler(EventHandler<Event<T>>... eventHandlers) {
if (this.handlerGroup != null) {
this.handlerGroup.handleEventsWith(eventHandlers);
} else {
this.handlerGroup = disruptor.handleEventsWith(eventHandlers);
}
return this;
}
public EventPublisher<T> workHandler(WorkHandler<Event<T>>... workHandlers) {
if (this.handlerGroup != null) {
this.handlerGroup.handleEventsWithWorkerPool(workHandlers);
} else {
this.handlerGroup = disruptor.handleEventsWithWorkerPool(workHandlers);
}
return this;
}
public EventPublisher<T> thenHandler(EventHandler<Event<T>>... thenHandlers) {
this.handlerGroup.then(thenHandlers);
return this;
}
public EventPublisher<T> afterHandler(EventHandler<Event<T>>... afterHandlers) {
this.handlerGroup = this.disruptor.after(afterHandlers);
return this;
}
public Disruptor<Event<T>> getDisruptor() {
return disruptor;
}
@Override
public EventPublisher<T> start() {
this.disruptor.start();
this.state = PublisherState.START;
return this;
}
@Override
public EventPublisher<T> shutdown() {
this.disruptor.shutdown();
this.state = PublisherState.SHUTDOWN;
return this;
}
@Override
public EventPublisher<T> publish(T t) {
if (!isStarted()) {
throw new IllegalStateException("publisher not start..");
}
if (producers == null || producers.length <= 0) {
throw new IllegalStateException("producer must be specify.");
}
for (Producer<T> producer : producers) {
producer.produce(t);
}
return this;
}
private boolean isStarted() {
return this.state == PublisherState.START;
}
}
其中，比较重要的代码是指定生产者及消费者链(！！！贼好用！！！)。动态参数表示生产者和消费者可以是一对一、一对多、多对一、多对多的关系。
先来讲讲publisher的最小数据单位Event
package com.gravel.demo.test.disruptor.base;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description:
*/
public class Event<T> {
private T data;
public void set(T data) {
this.data = data;
}
public T get() {
return data;
}
}
再来看看生产者的代码， EventProducer继承自Producer。
Producer.java
package com.gravel.demo.test.disruptor.base;
/**
* @Auther: syh
* @Date: 2020/7/9
* @Description:
*/
public interface Producer<T> {
void produce(T t);
}
EventProducer.java
package com.gravel.demo.test.disruptor.base;
import com.lmax.disruptor.RingBuffer;
/**
* @Auther: syh
* @Date: 2020/7/9
* @Description:
*/
public class EventProducer<T> implements Producer<T> {
private String name;
private RingBuffer<Event<T>> ringBuffer;
private EventTranslator<T> translator;
public EventProducer(String name) {
this.name = name;
}
public EventProducer<T> setRingBuffer(RingBuffer<Event<T>> ringBuffer) {
this.ringBuffer = ringBuffer;
return this;
}
public EventProducer<T> setTranslator(EventTranslator<T> translator) {
this.translator = translator;
return this;
}
@Override
public void produce(T t) {
System.out.println(String.format("producer message by %s, data: %s", name, t));
if (translator != null) {
ringBuffer.publishEvent(translator, t);
} else {
long seq = ringBuffer.next();
try {
Event<T> event = ringBuffer.get(seq);
event.set(t);
} finally {
// Disruptor 要求 RingBuffer.publish 必须得到调用的潜台词就是，如果发生异常也一样要调用 publish
// 如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer
ringBuffer.publish(seq);
}
}
}
}
其中，ringBuffer必须是Disruptor实例的成员。所以在指定producer时必须遍历设置ringBuffer。translator对象也做单例模式不知道是否会发生线程间数据覆盖问题。如果线程不安全，就每个Producer都初始化一个Translator对象。从produce()方法我们可以看见，disruptor是通过ringBuffer发布消息的。有两种发布方式：一种是通过translator方式，一种是通过sequence方式。注意finally里面的提示。
来看看EventTranslator的代码,继承自EventTranslatorVararg，translateTo()方法对EventFactory创建的实例做数据填充。
package com.gravel.demo.test.disruptor.base;
import com.lmax.disruptor.EventTranslatorVararg;
/**
* @Auther: syh
* @Date: 2020/7/9
* @Description:
*/
public class EventTranslator<T> implements EventTranslatorVararg<Event<T>> {
@Override
public void translateTo(Event<T> event, long sequence, Object... args) {
event.set(((T)(args[0])));
}
}
然后是消费者。为了统一处理worker和handler结果，所有的消费者必须实现Disruptor的EventHandler/WorkHandler。还有一个自定义的Consumer接口。
Consumer接口
package com.gravel.demo.test.disruptor.base;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description:
*/
public interface Consumer<T> {
void consume(T data, Boolean over) throws Exception;
}
EventConsumer.java
package com.gravel.demo.test.disruptor.base;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description:
*/
public abstract class EventConsumer<T> implements EventHandler<Event<T>>, Consumer<T>, WorkHandler<Event<T>> {
String name;
public EventConsumer(String name) {
this.name = name;
}
@Override
public void onEvent(Event<T> event, long seq, boolean over) throws Exception {
consume(event.get(), over);
}
@Override
public void onEvent(Event<T> event) throws Exception {
consume(event.get(), null);
}
protected String getName() {
return name;
}
}
抽象类EventConsumer统一处理worker和handler的onEvent方法。具体的consume操作需要用户自己实现。如demo里面的DomainConsumer。
到这里。比较关键的生产者和消费者代码都封装好了。
现在看看非核心代码
错误处理类：ErrorHandler实现ExceptionHandler
package com.gravel.demo.test.disruptor.base;
import com.lmax.disruptor.ExceptionHandler;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description:
*/
public abstract class ErrorHandler<T> implements ExceptionHandler<Event<T>> {
@Override
public void handleEventException(Throwable throwable, long l, Event<T> t) {
handle(t.get(), throwable);
}
@Override
public void handleOnStartException(Throwable throwable) {
}
@Override
public void handleOnShutdownException(Throwable throwable) {
}
protected abstract void handle(T object, Throwable throwable);
}
EventFactory事件工厂类，实现Disruptor的EeventFactory
package com.gravel.demo.test.disruptor.base;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description:
*/
public class EventFactory implements com.lmax.disruptor.EventFactory {
@Override
public Event newInstance() {
return new Event();
}
}
ringBuffer发送方式
package com.gravel.demo.test.disruptor.base;
/**
* @Auther: syh
* @Date: 2020/7/9
* @Description:
*/
public enum PublishStrategy {
NORMAL, TRANSLATOR
}
至此，Disruptor的封装就结束了。放一下demo类
实体Domain
package com.gravel.demo.test.disruptor;
/**
* @Auther: syh
* @Date: 2020/7/9
* @Description:
*/
public class Domain {
private String id;
private String value;
public Domain(String id, String value) {
this.id = id;
this.value = value;
}
public void setValue(String value) {
this.value = value;
}
@Override
public String toString() {
return "Domain{" +
"id=‘" + id + ‘\‘‘ +
", value=‘" + value + ‘\‘‘ +
‘}‘;
}
}
DomainConsumer
package com.gravel.demo.test.disruptor;
import com.gravel.demo.test.disruptor.base.EventConsumer;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description:
*/
public class DomainConsumer extends EventConsumer<Domain> {
public DomainConsumer() {
this("FirstDisruptorHandler" + (Math.random() * 100));
}
public DomainConsumer(String name) {
super(name);
}
@Override
public void consume(Domain data, Boolean over) throws Exception {
// errorHandler测试用：worker抛异常，handler正常处理
/*if (Objects.isNull(over)) {
throw new RuntimeException(getName() + " handle exception.");
}*/
System.out.println(String.format("received by %s, data: %s, is over?%s", getName(), data.toString(), over));
data.setValue(getName());
}
}
ConsumerErrorHandler
package com.gravel.demo.test.disruptor;
import com.gravel.demo.test.disruptor.base.ErrorHandler;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description:
*/
public class DomainErrorHandler<T> extends ErrorHandler<T> {
@Override
protected void handle(T object, Throwable throwable) {
System.err.println(String.format("received a error message: %s, data: %s, ", throwable.getMessage(), object));
// 不抛异常，则数据会继续流转到下一个handler
throw new IllegalStateException("interrupted.");
}
}
LimitThreadFactory
package com.gravel.demo.test.disruptor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @Auther: syh
* @Date: 2020/7/8
* @Description: 单线程
*/
public class LimitedThreadFactory implements ThreadFactory {
private final AtomicInteger count = new AtomicInteger(0);
public Thread newThread(Runnable r) {
if (count.compareAndSet(0, 1)) {
return new Thread(r);
} else {
throw new IllegalStateException("Created more that one thread");
}
}
}
demo测试（只测一条消费数据）
单生产单消费(worker和handler调用结果一致)
.producer(producer1) .handler(handler1)
调用结果
producer message by producer1, data: Domain{id=‘0‘, value=‘init‘}
received by handler1, data: Domain{id=‘0‘, value=‘init‘}, is over?true
used time: 19ms
单生产者多消费者(hanlder类型，设置菱形调用链。)。
.producer(producer1) .handler(handler1, handler2) .after(handler1).handler(after1) .after(handler2).handler(after2) .after(after1, after2).handler(after3)
调用结果(event数据会被每个handler都消费。)
producer message by producer1, data: Domain{id=‘0‘, value=‘init‘}
received by handler2, data: Domain{id=‘0‘, value=‘init‘}, is over?true
received by handler1, data: Domain{id=‘0‘, value=‘init‘}, is over?true
received by after2, data: Domain{id=‘0‘, value=‘handler1‘}, is over?true
received by after1, data: Domain{id=‘0‘, value=‘handler1‘}, is over?true
received by after3, data: Domain{id=‘0‘, value=‘after1‘}, is over?true
used time: 19ms
单线程 ，单生产者多消费者(worker类型，work类型不能直接设置after。)。
.producer(producer1) .worker(handler1, handler2)
调用结果（对比handler，可以看出event数据只会被一个worker消费。）
producer message by producer1, data: Domain{id=‘0‘, value=‘init‘}
received by handler1, data: Domain{id=‘0‘, value=‘init‘}, is over?null
used time: 21ms
多生产单消费（work和handler一致）
.handler(handler1) .producer(producer1, producer2)
运行结果(数据被改写了)
producer message by producer1, data: Domain{id=‘0‘, value=‘init‘}
producer message by producer2, data: Domain{id=‘0‘, value=‘init‘}
received by handler1, data: Domain{id=‘0‘, value=‘init‘}, is over?true
received by handler1, data: Domain{id=‘0‘, value=‘handler1‘}, is over?true
used time: 18ms
多生产多消费（worker和handler混合用)
.producer(producer1, producer2) .worker(after1, after2) .handler(handler1, handler2)
调用结果
producer message by producer1, data: Domain{id=‘0‘, value=‘init‘}
producer message by producer2, data: Domain{id=‘0‘, value=‘init‘}
received by after2, data: Domain{id=‘0‘, value=‘init‘}, is over?null
received by after1, data: Domain{id=‘0‘, value=‘after2‘}, is over?null
received by handler2, data: Domain{id=‘0‘, value=‘after1‘}, is over?false
received by handler1, data: Domain{id=‘0‘, value=‘after1‘}, is over?false
received by handler2, data: Domain{id=‘0‘, value=‘handler2‘}, is over?true
received by handler1, data: Domain{id=‘0‘, value=‘handler1‘}, is over?true
used time: 26ms
标签：dsl produce state sleep 覆盖 throws cer def 多个
原文地址：https://www.cnblogs.com/braska/p/13275289.html
