高性能队列框架-Disruptor使用、Netty结合Disruptor大幅提高数据处理性能
高性能队列框架-Disruptor
首先介绍一下 Disruptor 框架，Disruptor是一个通用解决方案，用于解决并发编程中的难题（低延迟与高吞吐量），Disruptor 在高并发场景下性能表现很好，如果有这方面需要，可以深入研究其源码
其本质还是一个队列（环形），与其他队列类似，也是基于生产者消费者模式设计，只不过这个队列很特别是一个环形队列。这个队列能够在无锁的条件下进行并行消费，也可以根据消费者之间的依赖关系进行先后次序消费。
使用 Disruptor 框架的好处就是：速度快！
生产者向 RingBuffer 写入，消费者从 RingBuffer 中消费，基于 Disruptor 开发的系统每秒可以支持 600 万订单
下边介绍一下 Disruptor 框架中常见概念：
RingBuffer
基于数组实现的一个环，用于在不同线程间传递数据，RingBuffer 有一个 Sequencer 序号器，指向数组中下一个可用元素
Sequencer 序号器
该类是 Disruptor 核心，有两个实现类：
- SingleProducerSequencer 单生产者
- MultiProducerSequencer 多生产者
WaitStrategy 等待策略
消费者等待生产者将数据放入 RingBuffer，有不同的等待策略：
BlockingWaitStrategy：阻塞等待策略，最低效的策略，但其对 CPU 的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现。SleepingWaitStrategy：休眠等待策略，性能表现跟 BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景。YieldingWaitStrategy：产生等待策略，性能最好，适合用于低延迟的系统，在要求极高性能且事件处理线程数小于 CPU 逻辑核心数的场景中，推荐使用。是无锁并行
Disruptor 的设计中是没有锁的，在 Disruptor 中出现线程竞争的地方也就是 RingBuffer 中的下标 Sequence，Disruptor 通过 CAS 操作来代替加锁，从而提升性能，CAS 的性能大约是加锁操作性能的 8 倍，
伪共享问题
Disruptor 中还会出现伪共享问题
参考：《高性能队列——Disruptor》——美团技术团队
缓存行
Cache 是由很多个 cache line 组成，每个 cache line 通常是 64B，并且可以有效地引用主内存中的一块地址。
Java 中 long 类型变量是 8B，因此一个 cache line 可以存储 8 个 long 类型变量
CPU 每次从主存中拉取数据时，会把相邻的数据也存入同一个 cache line，那么在访问一个 long 数组时，如果数组中的一个值被加入缓存中，那么也会加载另外 7 个
伪共享问题
在 ArrayBlockingQueue 中有 3 个成员变量：
- takeIndex：需要被取走元素下标
- putIndex：可被插入元素下标
- count：队列元素数量
这 3 个变量如果在同一个 cache line 中的话，假如此时有两个线程对这 3 个变量进行操作，线程 A 修改了 takeIndex 变量，那么会导致线程 B 中这个变量所在的 cache line 失效，需要从内存重新读取
这种无法充分利用 cache line 特性的线程，成为 伪共享
解决方案就是，增大数组元素之间的间隔，使得不同线程存取的元素位于不同的 cache line 上，通过空间换时间
在jdk1.8中，有专门的注解
@Contended来避免伪共享，更优雅地解决问题。
Disruptor 通过哪些设计来解决队列速度慢的问题了呢？
-
环形数组 RingBuffer
采用环形数组，空间重复利用，避免垃圾回收，并且数组对于缓存机制更加友好
-
元素位置定位
数组长度 2^n，通过位运算，加快定位速度
-
无锁设计
通过 CAS 代替锁来保证操作的线程安全
在美团内部，很多高并发场景借鉴了Disruptor的设计，减少竞争的强度。其设计思想可以扩展到分布式场景，通过无锁设计，来提升服务性能
Disruptor 多个生产者、多个消费者原理
在 Disruptor 中，多个生产者生产数据时，每个线程获取不同的一段数组空间再加上 CAS 操作，可以避免多个线程重复写同一个元素
在读取时，如何避免读取到未写的元素呢？
Disruptor 中新创建了一个与 RingBuffer 大小相同的 available Buffer，当某个位置写入成功，就在 available Buffer 中标记为 true，通过该标记来读取已经写好的元素
Disruptor 单生产者单消费者实战
首先引入依赖：
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.3.4</version>
</dependency>
定义订单：
/**
* 订单对象，生产者要生产订单对象，消费者消费订单对象
*/
public class OrderEvent {
// 订单的价格
private long value;
public long getValue() {
return value;
}
public void setValue(long value) {
this.value = value;
}
}
定义工厂类，用于创建订单对象：
/**
* 建立一个工厂类，用于创建Event的实例（OrderEvent)
*/
public class OrderEventFactory implements EventFactory<OrderEvent> {
@Override
public OrderEvent newInstance() {
// 生产对象
return new OrderEvent();
}
}
定义事件处理器，用于监听消费订单：
/**
* 消费者
*/
public class OrderEventHandler implements EventHandler<OrderEvent> {
@Override
public void onEvent(OrderEvent orderEvent, long l, boolean b) {
System.err.println("消费者:" + orderEvent.getValue());
}
}
定义生产者，用于生产订单：
public class OrderEventProducer {
// ringBuffer 用于存储数据
private RingBuffer<OrderEvent> ringBuffer;
public OrderEventProducer(RingBuffer<OrderEvent> ringBuffer) {
this.ringBuffer = ringBuffer;
}
// 生产者向 ringBuffer 中生产消息
public void sendData(ByteBuffer data) {
// 1. 生产者先从 ringBuffer 拿到可用的序号
long sequence = ringBuffer.next();
try {
// 2.根据这个序号找到具体的 OrderEvent 元素, 此时获取到的 OrderEvent 对象是一个没有被赋值的空对象
OrderEvent event = ringBuffer.get(sequence);
// 3. 设置订单价格
event.setValue(data.getLong(0));
} catch (Exception e) {
e.printStackTrace();
} finally {
// 4. 提交发布操作
ringBuffer.publish(sequence);
}
}
}
测试类：
public class Main {
public static void main(String[] args) {
// 初始化一些参数
OrderEventFactory orderEventFactory = new OrderEventFactory();
int ringBufferSize = 8;
ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
/**
* 参数说明：
* eventFactory:消息(event)工厂对象
* ringBufferSize: 容器的长度
* executor:线程池，建议使用自定义的线程池，线程上限。
* ProducerType:单生产者或多生产者
* waitStrategy:等待策略
*/
// 1. 实例化disruptor对象
Disruptor<OrderEvent> disruptor = new Disruptor<OrderEvent>(
orderEventFactory,
ringBufferSize,
executor,
ProducerType.SINGLE,
new BlockingWaitStrategy());
// 2. 向 Disruptor 中添加消费者，消费者监听到 Disruptor 的 RingBuffer 中有数据了，就会进行消费
disruptor.handleEventsWith(new OrderEventHandler());
// 3. 启动disruptor
disruptor.start();
// 4. 拿到存放数据的容器：RingBuffer
RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
// 5. 创建生产者
OrderEventProducer producer = new OrderEventProducer(ringBuffer);
// 6. 通过生产者向容器 RingBuffer 中存放数据
ByteBuffer bb = ByteBuffer.allocate(8);
for (long i = 0; i < 100; i++) {
bb.putLong(0, i);
producer.sendData(bb);
}
// 7.关闭
disruptor.shutdown();
executor.shutdown();
}
}
Disruptor 多生产者和多消费者实战
定义消费者，用于从 ringBuffer 中消费订单：
public class ConsumerHandler implements WorkHandler<Order> {
// 每个消费者有自己的id
private String comsumerId;
// 计数统计，多个消费者，所有的消费者总共消费了多个消息。
private static AtomicInteger count = new AtomicInteger(0);
private Random random = new Random();
public ConsumerHandler(String comsumerId) {
this.comsumerId = comsumerId;
}
// 当生产者发布一个 sequence，ringbuffer 中一个序号，里面生产者生产出来的消息，生产者最后publish发布序号
// 消费者会监听，如果监听到，就会ringbuffer去取出这个序号，取到里面消息
@Override
public void onEvent(Order event) throws Exception {
// 模拟消费者处理消息的耗时，设定1-4毫秒之间
TimeUnit.MILLISECONDS.sleep(1 * random.nextInt(5));
System.out.println("当前消费者:" + this.comsumerId + ", 消费信息 ID:" + event.getId());
// count 计数器增加 +1，表示消费了一个消息
count.incrementAndGet();
}
// 返回所有消费者总共消费的消息的个数。
public int getCount() {
return count.get();
}
}
定义订单：
@Data
public class Order {
private String id;
private String name;
private double price;
public Order() {
}
}
定义生产者，用于向 ringBuffer 中生产订单：
public class Producer {
private RingBuffer<Order> ringBuffer;
// 为生产者绑定 ringBuffer
public Producer(RingBuffer<Order> ringBuffer) {
this.ringBuffer = ringBuffer;
}
// 发送数据
public void sendData(String uuid) {
// 1. 获取到可用sequence
long sequence = ringBuffer.next();
try {
Order order = ringBuffer.get(sequence);
order.setId(uuid);
} finally {
// 2. 发布序号
ringBuffer.publish(sequence);
}
}
}
测试类：
public class TestMultiDisruptor {
public static void main(String[] args) throws InterruptedException {
// 1. 创建 RingBuffer，Disruptor 包含 RingBuffer
RingBuffer<Order> ringBuffer = RingBuffer.create(ProducerType.MULTI, // 多生产者
new EventFactory<Order>() {
@Override
public Order newInstance() {
return new Order();
}
}, 1024 * 1024, new YieldingWaitStrategy());
// 2. 创建 ringBuffer 屏障
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
// 3. 创建多个消费者数组
ConsumerHandler[] consumers = new ConsumerHandler[10];
for (int i = 0; i < consumers.length; i++) {
consumers[i] = new ConsumerHandler("C" + i);
}
// 4. 构建多消费者工作池
WorkerPool<Order> workerPool = new WorkerPool<Order>(ringBuffer, sequenceBarrier, new EventExceptionHandler(), consumers);
// 5. 设置多个消费者的 sequence 序号，用于单独统计消费者的消费进度。消费进度让RingBuffer知道
ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
// 6. 启动 workPool
workerPool.start(Executors.newFixedThreadPool(5)); // 在实际开发，自定义线程池。
final CountDownLatch latch = new CountDownLatch(1);
// 100 个生产者向 ringBuffer 生产数据，每个生产者发送 100 个数据，共 10000 个数据
for (int i = 0; i < 100; i ++) {
final Producer producer = new Producer(ringBuffer);
new Thread(new Runnable() {
@Override
public void run() {
try {
// 先等待创建完 100 个生产者之后，再发送数据
latch.await();
} catch (Exception e) {
e.printStackTrace();
}
// 每个生产者发送 100 个数据
for (int j = 0; j < 100; j ++) {
producer.sendData(UUID.randomUUID().toString());
}
}
}).start();
}
// 把所有线程都创建完
TimeUnit.SECONDS.sleep(2);
// 唤醒线程让生产者开始发送数据，开始运行100个线程
latch.countDown();
// 等待数据发送完毕
TimeUnit.SECONDS.sleep(10);
System.out.println("任务总数:" + consumers[0].getCount());
}
static class EventExceptionHandler implements ExceptionHandler<Order> {
//消费时出现异常
@Override
public void handleEventException(Throwable throwable, long l, Order order) {
}
//启动时出现异常
@Override
public void handleOnStartException(Throwable throwable) {
}
//停止时出现异常
@Override
public void handleOnShutdownException(Throwable throwable) {
}
}
}
Disruptor 与 Netty 结合大幅提高数据处理性能
使用 Netty 接收处理数据时，不要在工作线程上进行处理，降低 Netty 性能，可以使用异步机制，通过线程池来处理，异步处理的话，就是用 Disruptor 来作为任务队列即可
即在 Netty 收到处理数据请求时，封装成一个事件，向 Disruptor 中推送，再通过多消费者来进行处理，可以提升 Netty 处理数据时的性能，流程图如下（绿色部分为通过 Disruptor 优化部分）：
