- Java锁的逻辑(结合对象头和ObjectMonitor)
- 还在用饼状图?来瞧瞧这些炫酷的百分比可视化新图形(附代码实现)⛵
- 自动注册实体类到EntityFrameworkCore上下文,并适配ABP及ABPVNext
- 基于Sklearn机器学习代码实战
介绍高性能队列Disruptor原理以及使用例子.
Disruptor是外汇和加密货币交易所运营商 LMAX group 建立高性能的金融交易所的结果。用于解决生产者、消费者及其数据存储的设计问题的高性能队列实现。可以对标JDK中的ArrayBlockingQueue。是目前单机且基于内存存储的最高性能的队列实现。见 与ArrayBlockingQueue性能对比 .
锁非常昂贵,因为它们在竞争时需要仲裁。这种仲裁是通过到操作系统内核的上下文切换来实现的,该内核将挂起等待锁的线程,直到它被释放。系统提供的原子操作CAS(Compare And Swap/Set)是很好的锁替代方案,Disruptor中同步就是使用的这种.
比如多生产者模式中com.lmax.disruptor.MultiProducerSequencer就是用了Java里sun.misc.Unsafe类基于CAS实现的API.
等待策略com.lmax.disruptor.BlockingWaitStrategy使用了基于CAS实现的ReentrantLock.
为了提高效率CPU硬件不会以字节或字为单位移动内存,而是以缓存行,通常大小为 32-256 字节的缓存行,最常见的缓存行是 64 字节。这意味着,如果两个变量在同一个缓存行中,并且由不同的线程写入,那么它们会出现与单个变量相同的写入争用问题。为了获得高性能,如果要最小化争用,那么确保独立但同时写入的变量不共享相同的缓存行是很重要的.
比如com.lmax.disruptor.RingBuffer中属性前后都用未赋值的long来独占。com.lmax.disruptor.SingleProducerSequencerPad也有相同处理方式.
队列相比链表在访问速度上占据优势,而有界队列相比可动态扩容的无界队列则避免扩容产生的同步问题效率更高。Disruptor和JDK中的ArrayBlockingQueue一样使用有界队列。队列长度要设为2的n次幂,有利于二进制计算.
Disruptor在逻辑上将数组的的头尾看成是相连的,即一个环形数组(RingBuffer).
生产和消费都需要维护自增序列值(Sequence),从0开始.
生产方只维护一个代表生产的最后一个元素的序号。代表生产的最后一个元素的序号。每次向Disruptor发布一个元素都调用Sequenced.next()来获取下个位置的写入权.
在单生产者模式(SINGLE)由于不存在并发写入,则不需要解决同步问题。在多生产者模式(MULTI)就需要借助JDK中基于CAS(Compare And Swap/Set)实现的API来保证线程安全.
多个消费者各自维护自己的消费序列值(Sequence)保存数组中.
而环形通过与运算(sequence & indexMask)实现的,indexMask就是环形队列的长度-1。以环形队列长度8为例,第9个元素Sequence为8,8 & 7 = 0,刚好又回到了数组第1个位置.
见com.lmax.disruptor.RingBuffer.elementAt(long sequence) 。
环形队列存放的是Event对象,而且是在Disruptor创建的时候调用EventFactory创建并一次将队列填满。Event保存生产者生产的数据,消费也是通过Event获取,后续生产则只需要替换掉Event中的属性值。这种方式避免了重复创建对象,降低JVM的GC产频率.
见com.lmax.disruptor.RingBuffer.fill(EventFactory eventFactory) 。
当消费速度大于生产速度情况下,消费者执行的等待策略.
策略类名 | 描述 |
---|---|
BlockingWaitStrategy(常用) | 使用ReentrantLock,失败则进入等待队列等待唤醒重试。当吞吐量和低延迟不如CPU资源重要时使用。 |
YieldingWaitStrategy(常用) | 尝试100次,全失败后调用Thread.yield()让出CPU。该策略将使用100%的CPU,如果其他线程请求CPU资源,这种策略更容易让出CPU资源。 |
SleepingWaitStrategy(常用) | 尝试200次 。前100次直接重试,后100次每次失败后调用Thread.yield()让出CPU,全失败线程睡眠(默认100纳秒 )。 |
BusySpinWaitStrategy | 线程一直自旋等待,比较耗CPU。最好是将线程绑定到特定的CPU核心上使用。 |
LiteBlockingWaitStrategy | 与BlockingWaitStrategy类似,区别在增加了原子变量signalNeeded,如果两个线程同时分别访问waitFor()和signalAllWhenBlocking(),可以减少ReentrantLock加锁次数。 |
LiteTimeoutBlockingWaitStrategy | 与LiteBlockingWaitStrategy类似,区别在于设置了阻塞时间,超过时间后抛异常。 |
TimeoutBlockingWaitStrategy | 与BlockingWaitStrategy类似,区别在于设置了阻塞时间,超过时间后抛异常。 |
PhasedBackoffWaitStrategy | 根据时间参数和传入的等待策略来决定使用哪种等待策略。当吞吐量和低延迟不如CPU资源重要时,可以使用此策略。 |
所有消费者的消费序列(Sequence)都放在一个数组中,见com.lmax.disruptor.AbstractSequencer,通过SEQUENCE_UPDATER来更新对应的序列值.
调用更新的地方在com.lmax.disruptor.RingBuffer.addGatingSequences(Sequence... gatingSequences).
生产者线程被阻塞。生产者调用Sequenced.next()争夺写入权的时候需要判断最小的消费序列值进行比较。如果写入的位置还未消费则会进入循环不断获取最小消费序列值进行比较.
见包com.lmax.disruptor下SingleProducerSequencer或MultiProducerSequencer中next(int n)方法.
Event是环形队列(RingBuffer)中的元素,是生产者数据的载体;EventFactory是定义Event创建方式的工厂类;EventHandler则是Event的处理器,定义如何消费Event中的数据.
另外有必要定义一个消费异常处理器ExceptionHandler,它是和EventHandler绑定的。当EventHandler.onEvent()执行抛出异常时会执行对应的异常回调方法.
创建Disruptor需要指定5个参数eventFactory、ringBufferSize、threadFactory、producerType、waitStrategy.
EventFactory是上面定义的Event工厂类; 。
ringBufferSize是环形队列的长度,这个值要是2的N次方; 。
threadFactory是定义消费者线程创建方式的工厂类; 。
producerType是指明生产者是一个(SINGLE)还是多个(MULTI)。默认是MULTI,会使用CAS(Compare And Swap/Set)保证线程安全。如果指定为SINGLE,则不使用没必要的CAS,使单线程处理更高效.
waitStrategy指明消费者等待生产时的策略.
指明EventHandler并绑定ExceptionHandler。指定多个EventHandler时,会为每个EventHandler分配一个线程,一个Event会被多个并行EventHandler处理.
也可以指明多个WorkHandler,每个WorkHandler分配一个线程并行消费队列中的Event,一个Event只会被一个WorkHandler处理.
EventTranslator定义生产者数据转换为Event的方式,不同数量参数有不同的接口用来实现.
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.4.4</version>
</dependency>
/**
* 事件
*
* @param <T>发布的数据类型
*/
public class MyEvent<T> {
private T data;
public T getData() {
return data;
}
public MyEvent<T> setData(T data) {
this.data = data;
return this;
}
}
import com.lmax.disruptor.EventFactory;
/**
* 创建事件的工厂
*
* @param <T>发布的数据类型
*/
public class MyEventFactory<T> implements EventFactory<MyEvent<T>> {
@Override
public MyEvent<T> newInstance() {
return new MyEvent<>();
}
}
import com.lmax.disruptor.EventHandler;
/**
* 事件消费方法
*
* @param <T>发布的数据类型
*/
public class MyEventHandler<T> implements EventHandler<MyEvent<T>> {
@Override
public void onEvent(MyEvent<T> tMyEvent, long l, boolean b) throws Exception {
System.out.println(Thread.currentThread().getName() + "MyEventHandler消费:" + tMyEvent.getData());
}
}
import com.lmax.disruptor.ExceptionHandler;
/**
* 消费者异常处理器
*
* @param <T>发布的数据类型
*/
public class MyExceptionHandler<T> implements ExceptionHandler<MyEvent<T>> {
@Override
public void handleEventException(Throwable ex, long sequence, MyEvent<T> event) {
System.out.println("handleEventException");
}
@Override
public void handleOnStartException(Throwable ex) {
System.out.println("handleOnStartException");
}
@Override
public void handleOnShutdownException(Throwable ex) {
System.out.println("handleOnShutdownException");
}
}
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import static com.lmax.disruptor.dsl.ProducerType.SINGLE;
/**
* 单消费者
*/
public class SingleConsumerSample {
public static void main(String[] args) {
// 环形数组长度,必须是2的n次幂
int ringBufferSize = 1024;
// 创建事件(Event)对象的工厂
MyEventFactory<String> eventFactory = new MyEventFactory<>();
// 创建消费者线程工厂
ThreadFactory threadFactory = Executors.defaultThreadFactory();
// 等待策略
WaitStrategy waitStrategy = new SleepingWaitStrategy();
Disruptor<MyEvent<String>> disruptor =
new Disruptor<>(eventFactory, ringBufferSize, threadFactory, SINGLE, waitStrategy);
// 指定一个处理器
MyEventHandler<String> eventHandler = new MyEventHandler<>();
disruptor.handleEventsWith(eventHandler);
// 处理器异常处理器
ExceptionHandler<MyEvent<String>> exceptionHandler = new MyExceptionHandler<>();
disruptor.setDefaultExceptionHandler(exceptionHandler);
disruptor.start();
// 通过事件转换器(EventTranslator)来指明如何将发布的数据转换到事件对象(Event)中
// 这里是一个参数的转换器,另外还有两个(EventTranslatorTwoArg)、三个(EventTranslatorThreeArg)
// 和多个(EventTranslatorVararg)参数的转换器可以使用,参数类型可以不一样
EventTranslatorOneArg<MyEvent<String>, String> eventTranslatorOneArg =
new EventTranslatorOneArg<MyEvent<String>, String>() {
@Override
public void translateTo(MyEvent<String> event, long sequence, String arg0) {
event.setData(arg0);
}
};
// 发布
for (int i = 0; i < 10; i++) {
disruptor.publishEvent(eventTranslatorOneArg, "One arg " + i);
}
disruptor.shutdown();
}
}
这种只是迎合Java8 Lambda语法特性,代码更简洁.
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.stream.Collectors;
import static com.lmax.disruptor.dsl.ProducerType.SINGLE;
public class LambdaSample {
public static void main(String[] args) {
// 环形数组长度,必须是2的n次幂
int ringBufferSize = 1024;
// 创建消费者线程工厂
ThreadFactory threadFactory = Executors.defaultThreadFactory();
// 等待策略
WaitStrategy waitStrategy = new SleepingWaitStrategy();
Disruptor<MyEvent<String>> disruptor =
new Disruptor<>(MyEvent::new, ringBufferSize, threadFactory, SINGLE, waitStrategy);
// 指定一个处理器
EventHandler<MyEvent<String>> eventHandler = (event, sequence, endOfBatch) ->
System.out.println(Thread.currentThread().getName() + "MyEventHandler消费:" + event.getData());
disruptor.handleEventsWith(eventHandler);
// 处理器异常处理器
ExceptionHandler<MyEvent<String>> exceptionHandler = new MyExceptionHandler<>();
disruptor.setDefaultExceptionHandler(exceptionHandler);
disruptor.start();
// 通过事件转换器(EventTranslator)来指明如何将发布的数据转换到事件对象(Event)中
// 一个参数的转换器
disruptor.publishEvent((event, sequence, param) -> event.setData(param), "One arg ");
// 两个参数的转换器
disruptor.publishEvent((event, sequence, pA, pB) -> event.setData(pA + pB), "Two arg ", 1);
// 三个参数的转换器
disruptor.publishEvent((event, sequence, pA, pB, pC) -> event.setData(pA + pB + pC)
, "Three arg ", 1, false);
// 多个参数的转换器
disruptor.getRingBuffer().publishEvent((event, sequence, params) -> {
List<String> paramList = Arrays.stream(params).map(Object::toString).collect(Collectors.toList());
event.setData("Var arg " + String.join(",", paramList));
}, "param1", "param2", "param3");
disruptor.shutdown();
}
}
关键只在于指定多个EventHandler,并且EventHandler还可以分别绑定不同的ExceptionHandler.
每个EventHandler分配一个线程,一个Event会被每个EventHandler处理,适合两个不同的业务都需要处理同一个元素的情况,类似广播模式.
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import static com.lmax.disruptor.dsl.ProducerType.SINGLE;
/**
* 一个元素多个消费者重复消费
*/
public class RepetitionConsumerSample {
public static void main(String[] args) {
// 环形数组长度,必须是2的n次幂
int ringBufferSize = 1024;
// 创建事件(Event)对象的工厂
MyEventFactory<String> eventFactory = new MyEventFactory<>();
// 创建消费者线程工厂
ThreadFactory threadFactory = Executors.defaultThreadFactory();
// 等待策略
WaitStrategy waitStrategy = new SleepingWaitStrategy();
Disruptor<MyEvent<String>> disruptor =
new Disruptor<>(eventFactory, ringBufferSize, threadFactory, SINGLE, waitStrategy);
// 这里指定了2个消费者,那就会产生2个消费线程,一个事件会被消费2次
EventHandler<MyEvent<String>> eventHandler = (event, sequence, endOfBatch) ->
System.out.println(Thread.currentThread().getName() + "MyEventHandler消费:" + event.getData());
EventHandler<MyEvent<String>> eventHandler2 = (event, sequence, endOfBatch) ->
System.out.println(Thread.currentThread().getName() + "MyEventHandler——2消费:" + event.getData());
disruptor.handleEventsWith(eventHandler, eventHandler2);
// 分别指定异常处理器
ExceptionHandler<MyEvent<String>> exceptionHandler = new MyExceptionHandler<>();
disruptor.handleExceptionsFor(eventHandler).with(exceptionHandler);
disruptor.handleExceptionsFor(eventHandler2).with(exceptionHandler);
disruptor.start();
for (int i = 0; i < 10; i++) {
disruptor.publishEvent((event, sequence, param) -> event.setData(param), "One arg " + i);
}
disruptor.shutdown();
}
}
关键只在于定义WorkHandler,然后实例化多个来消费.
每个WorkHandler分配一个线程,一个元素只会被一个WorkHandler处理.
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import static com.lmax.disruptor.dsl.ProducerType.SINGLE;
public class MultiConsumerSample {
public static void main(String[] args) {
// 环形数组长度,必须是2的n次幂
int ringBufferSize = 1024;
// 创建事件(Event)对象的工厂
MyEventFactory<String> eventFactory = new MyEventFactory<>();
// 创建消费者线程工厂
ThreadFactory threadFactory = Executors.defaultThreadFactory();
// 等待策略
WaitStrategy waitStrategy = new SleepingWaitStrategy();
Disruptor<MyEvent<String>> disruptor =
new Disruptor<>(eventFactory, ringBufferSize, threadFactory, SINGLE, waitStrategy);
// 处理器异常处理器
ExceptionHandler<MyEvent<String>> exceptionHandler = new MyExceptionHandler<>();
disruptor.setDefaultExceptionHandler(exceptionHandler);
// 设置2个消费者,2个线程,一个Event只被一个消费者消费
WorkHandler<MyEvent<String>> workHandler = tMyEvent ->
System.out.println(Thread.currentThread().getName() + "WorkHandler消费:" + tMyEvent.getData());
disruptor.handleEventsWithWorkerPool(workHandler, workHandler2);
disruptor.start();
for (int i = 0; i < 10; i++) {
disruptor.publishEvent((event, sequence, param) -> event.setData(param), "One arg " + i);
}
disruptor.shutdown();
}
}
Disruptor 主页 。
Disruptor 技术文档 。
GitHub Disruptor 。
GitHub Disruptor Getting Started 。
Maven Repository Disruptor Framework 。
LMAX 官网 。
最后此篇关于单机最快的队列Disruptor解析和使用的文章就讲到这里了,如果你想了解更多关于单机最快的队列Disruptor解析和使用的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。
如何制作 agario 的 java 单人游戏版本?我想使用面向对象编程。所以应该有 Dot、Blob 和一个 Controller Object。(我只是想分享我的代码,所以不要投反对票:) 最佳答
我有一个关于速度和性能的问题在单台机器上使用多个虚拟化节点 VS 在单台机器上使用单节点。 哪一个会表现更好? 之所以问这个问题是因为我目前是在单机上学习hadoop,在网上看到一些教程是单机多虚拟化
语境 我喜欢 Clojure。我喜欢 Clojure 胜过喜欢 ClojureScript。 我喜欢 DOM 胜过喜欢 Swing。 我打算主要在 Clojure 中构建一个 GUI 应用程序,但使用
我们目前正在使用 Opencart - 2.3.x 通过 ELB 为一个 AWS 实例下的多个域提供服务。 我们已经为一个域设置了 HTTPS,但对于其他域,我们不能做同样的事情,因为我们不能创建重复
我是一名优秀的程序员,十分优秀!