图解Disruptor框架（二）：核心概念

zoukankan html css js c++ java

图解Disruptor框架（二）：核心概念
图解Disruptor框架（二）：核心概念


概述

上一个章节简单的介绍了了下Disruptor，这节就是要好好的理清楚Disruptor中的核心的概念。并且会给出个HelloWorld的小例子。
在正式开始之前，我觉得有一点我感触非常的深刻，那就是：外国人取的类名真的真的非常的合适以及形象！看看接下来的内容就知道了！

核心概念介绍

下面这张图，非常好的总结了Disruptor中需要了解的核心概念：

多生产者多消费者模型

RingBuffer: Disruptor中的数据结构，用于存储生产者生产的数据（在Disruptor中，叫做Event）。

Sequence：序号。在Disruptor框架中，任何地方都有序号。生产者生产的数据放在RingBuffer中的哪个位置，消费者应该消费哪个位置的数据，RingBuffer中的某个位置的数据是什么，这些都是由这个序号来决定的。这个序号可以简单的理解为一个AtomicLong类型的变量。其使用了padding的方法去消除缓存的伪共享问题。

Sequencer:序号生成器。这个类主要是用来协调生产者的。在生产者生产数据的时候，Sequencer会产生一个可用的序号（Sequence），然后生产者就乖乖的把数据放在那里了。（此处不严谨，后续会说明原因。）

SequencerBarrier：序号屏障。（我觉得这个名字真的太形象了！）我们都知道，消费者在消费数据的时候，需要知道消费哪个位置的数据。消费者总不能自己想取哪个数据消费，就取哪个数据消费吧。（这样多混乱啊！）这个SequencerBarrier起到的就是这样一个“栅栏”般的阻隔作用。你消费者想消费数据，得，我告诉你一个序号（Sequence），你去消费那个位置上的数据。要是没有数据，就好好等着吧（怎么等也是有讲究的）。

先小结一下：Sequence、Sequencer、SequencerBarrier这三个概念一开始我是比较难以理解的。但是总结一下，无非就是Ringbuffer中哪里都需要用到序号，而Sequencer用于生产者生产的时候产生序号，SeqencerBarrier就是协调生产者与消费者，并且告诉消费者一个可以用于消费的序号！

Wait Strategy：等待策略。设想一种这样的情景：生产者生产的非常慢，而消费者消费的非常快。那么必然会出现数据不够的情况，这个时候消费者怎么进行等待呢？WaitStrategy就是为了解决问题而诞生的。

Event：数据、事件。这个Event就是我们希望RingBuffer存储的数据类型。这个是我们用户自己定义的。我们可以定义为任何事情，比如处理订单、消息等等。

EventHandler：事件处理器。当RingBuffer中有数据的时候，消费者怎么去对数据进行处理，就是由这个类去决定的。

Producer：生产者。用于生产数据。

上述就是Disruptor框架中的核心概念。如果不是非常的理解，可以先跟着我下面的例子去手敲一遍入门的程序，去体验一下。然后看看源码加深理解。

第一个入门小程序

首先是订单类Order

@Data
public class Order {

/**
* 订单ID
*/
private String id;

/**
* 订单名字
*/
private String name;

/**
* 用于记录这个对象创建的时间
*/
private Date createTime;
}

订单工厂类OrderFactory。

public class OrderFactory implements EventFactory<Order> {
public Order newInstance() {
Order order = new Order();
order.setCreateTime(new Date());
return order;
}
}

这个类主要用于ringbuffer构造的时候对其中存放的数据进行预加载。
相应的源码如下面所示：

/**
* 用于构造的时候需加载数据
* @param eventFactory
*/
private void fill(EventFactory<E> eventFactory)
{
for (int i = 0; i < bufferSize; i++)
{
// entries数组就是ringbuffer中用于存放数据的数组
entries[BUFFER_PAD + i] = eventFactory.newInstance();
}
}

生产者Producer

public class Producer {

private RingBuffer<Order> ringBuffer;

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

public void sendData(String s) {
long sequence = ringBuffer.next();
Order order = ringBuffer.get(sequence);
order.setId(s);
ringBuffer.publish(sequence);
}
}

消费者Comsumer

@Data
public class Comsumer implements WorkHandler<Order> {

/**
* 消费者ID
*/
private String comsumerId;

/**
* 记录消费的次数
*/
public static final AtomicInteger count = new AtomicInteger(0);

public Comsumer(String comsumerId) {
this.comsumerId = comsumerId;
}

@Override
public void onEvent(Order event) throws Exception {
System.out.println("消费者："+comsumerId+"消费了数据，ID="+event.getId()+"Name="+event.getName());
count.incrementAndGet();
}
}

主函数Main

public class Main {
public static void main(String[] args) throws InterruptedException {

// 构造ringbuffer
RingBuffer<Order> ringBuffer = RingBuffer.create(
ProducerType.MULTI,
new OrderFactory(),
1024*1024,
new BlockingWaitStrategy()
);

// 创建一个序号屏障
SequenceBarrier barrier = ringBuffer.newBarrier();

// 创建消费者数组
Comsumer[] comsumers = new Comsumer[10];
for (int i = 0; i < comsumers.length; i++) {
comsumers[i] = new Comsumer("comsumer"+i);
}

// 创建多消费者的工作池
WorkerPool<Order> workerPool = new WorkerPool<Order>(
ringBuffer,
barrier,
new EventExceptionHandler(),
comsumers
);

// 设置多个消费者的sequence序号用于单独统计消费进度, 并且设置到ringbuffer中
ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

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

// -----消费者创建完毕-----

// -----创建生产者------
// 用于阻塞生产者生产数据
CountDownLatch latch = new CountDownLatch(1);

for(int i = 0; i < 100; i++) {
final Producer producer = new Producer(ringBuffer);
new Thread(new Runnable() {
public void run() {
try {
// 阻塞生产者
latch.await();
} catch (Exception e) {
e.printStackTrace();
}
for(int j = 0; j<100; j++) {
producer.sendData(UUID.randomUUID().toString());
}
}
}).start();
}

System.err.println("----------线程创建完毕，开始生产数据----------");
// 放行生产者，让其生产数据
latch.countDown();

// 等待消费者消费完数据
Thread.sleep(2000);

System.err.println("消费次数："+Comsumer.count);

}

static class EventExceptionHandler implements ExceptionHandler<Order> {
public void handleEventException(Throwable ex, long sequence, Order event) {
}

public void handleOnStartException(Throwable ex) {
}

public void handleOnShutdownException(Throwable ex) {
}
}
}

运行结果：

上述的代码主要就是生产者生产了10000个数据，然后消费者再消费这些数据。可以看到结果是正确的。

上述这个小小的demo就结束了。接下来会再看看Disruptor的操作，例如链式操作、菱形操作、多边形操作等等。

总结

这一个小节，阐述了Disruptor中的一些核心概念，并编写了一个helloworld程序。如果读者觉得还是比较难理解，那就多敲几遍。这个没有办法的。
项目源码地址： https://gitee.com/cjh95/disruptor_blog/tree/master

参考资料

官网的简单介绍 https://github.com/LMAX-Exchange/disruptor/wiki/Introduction

伪共享 https://www.cnblogs.com/cyfonly/p/5800758.html
查看全文

相关阅读:
delphi xe10 FMX 启动参数
 delphi xe6 JSON 测试
 oracle实现http请求,oracle发送http请求。
ORACLE存储过程调用Web Service
新搭建的iis服务器，运行网站报 System.BadImageFormatException：未能加载文件或程序集”....“或它的某一个依赖项。
c#的http请求工具类核心代码
 vue-cli3 取消关闭eslint 校验代码
 quartz.net数据库持久化教程
 sql备份一张表的数据
 iis 长期无访问导致定时任务不执行的解决方案

原文地址：https://www.cnblogs.com/junhong1995/p/10146399.html

图解Disruptor框架（二）：核心概念

图解Disruptor框架（二）：核心概念

概述

核心概念介绍

第一个入门小程序

总结

参考资料