编程学习程序员架构算法设计模式和编程理论

LMAX Disruptor简介

2017-02-04  本文已影响383人  零度沸腾_yjz

LMAX是什么？

要说Disruptor需要先说下LMAX,LMAX是一个英国外汇黄金交易所，它是第一家也是唯一一家采用多边交易设施Multilateral Trading Facility(MTF)，拥有交易所拍照和经纪商拍照的欧洲顶级金融公司。而LMAX所用的Disruptor技术，在一个线程每秒处理6百万订单。没错，这个Disruptor就是我们这里的
Disruptor。
而Disruptor只是LMAX平台一部分，LMAX是一个新型零售金融交易平台，它能够达到低延迟、高吞吐量(大量交易)。这个系统建立在JVM平台上，核心是一个逻辑处理器，每秒能够处理600百万订单。业务逻辑处理器完全运行在内存中(in-memory)，使用事件源驱动方式(event sourcing)。而业务逻辑处理器核心是Disruptor，这是一个并发组件，能够在无锁情况下实现网络并发查询操作。他们研究表明，现在所谓的高性能研究方向似乎和现在CPU设计是相左的。

什么是Disruptor

Disruptor

Disruptor实现了队列的功能，而且是一个有界的队列。所以应用场景自然就是"生产者-消费者"模型了。可以看下JDK中的BlockingQuery是一个FIFO队列，生产者(Producer)发布(Publish)一项事件(Event，消息)时，消费者(Consumer)能够获得通知；当队列中没有事件时，消费者会被阻塞，直到生产者发布了新的事件。而Disruptor不仅仅只是这些：

可能你对这种场景还不是很明白，简单说就是当需要两个独立的处理过程(两个线程)之间需要传递数据时，就可以使用Disruptor，当然可以使用队列。

Disruptor中的核心概念

Models

简单Demo

1.定义事件

事件(Event)是Disruptor进行数据交换的数据类型

<pre>
public class PeopleEvent {
private String name;
private Integer age;
private Integer sex;

public void setName(String name) {
    this.name = name;
}

public String getName() {
    return name;
}

public void setAge(Integer age) {
    this.age = age;
}

public void setSex(Integer sex) {
    this.sex = sex;
}

public Integer getAge() {
    return age;
}

public Integer getSex() {
    return sex;
}

}
</pre>

2.定义事件工厂

事件工厂(Event Factory)用来实例化之前的事件(Event)，需要实现接口com.lmax.disruptor.EventFactory<>。Disruptor通过EventFactory在RingBuffer中创建Event的实例。一个Event实例实际被用作一个"数据槽"，发布者发布前，先从RingBuffer获得一个Event的实例，然后往Event中填充数据，之后发布到RingBuffer中，之后由Consumer获得该Event实例并从中取出数据。

<pre>
public class PeopleEventFactory implements EventFactory<PeopleEvent> {
public PeopleEvent newInstance(){
return new PeopleEvent();
}
}
</pre>

3.定义事件处理的具体实现(业务逻辑核心)
需要实现com.lmax.disruptor.EventHandler<>接口，来定义事件处理的具体逻辑

<pre>
public class PeopleEventHandler implements EventHandler<PeopleEvent> {

public void onEvent(PeopleEvent event, long sequence, boolean endOfBatch) throws Exception {
    System.out.println("name:" + event.getName()+",sex:" + event.getSex() + ",age:" + event.getAge());
}

}
</pre>

4.组合事件处理流程

<pre>
public class DisruptorDemo {

public static void main(String[] args){

    //Executor将用来为消费者构建线程
    Executor executor = Executors.newCachedThreadPool();

    //事件工厂用来创建事件
    PeopleEventFactory peopleEventFactory = new PeopleEventFactory();

    //指定Ring Buffer大小，2的倍数
    int buffSize = 1024;

    /**
     * 构造Disruptor
     * 并发系统提高性能之一就是单一写者原则，如果代码中仅有一个事件生产者，可以设置单一生产者模式来提高系统的性能。
     * 通过ProduceType.SINGLE和ProduceType.MULTI进行控制。
     *
     * 等待策略
     * Disruptor默认的等待策略是BlockingWaitStrategy，使用一个锁和条件变量来控制执行和等待，这是最慢的策略，但也是CPU使用最低
     * 和最稳定的策略。
     * SleepingWaitStrategy：也是CPU使用率低的策略，它使用循环等待并且循环间调用LockSupport.parkNanos(1)来睡眠。它的优点在于
     * 生产线程只需记数，而不执行任何命令，并且没有条件变量的消耗。但是对象从生产者到消费者传递延迟变大了，适用于不需要低延迟的场景，
     * YieldingWaitStrategy：是可以被用作低延迟系统的两个策略之一，这种策略在低延迟同时会增加CPU运算量。YieldingWaitStrategy
     * 会循环等待sequence增加到合适值，循环调用Tread.yield()允许其它准备好的线程执行。如果高性能而且事件消费者线程比逻辑内核少的
     * 时候，推荐使用YieldingWaitStrategy策略。
     * BusySpinWaitStrategy是性能最高的策略，同时也是对部署环境要求最高的策略。这个策略最好用在时间处理线程比物理内核数目还要少的时候。
     */
    Disruptor<PeopleEvent> disruptor = new Disruptor<PeopleEvent>(peopleEventFactory,buffSize,executor,
            ProducerType.SINGLE,new YieldingWaitStrategy());

    //链接处理器
    disruptor.handleEventsWith(new PeopleEventHandler());

    //启动Disruptor，启动所有线程
    disruptor.start();

    //从Disruptor获取RingBuffer，用来发布
    RingBuffer<PeopleEvent> ringBuffer = disruptor.getRingBuffer();

    PeopleEventProducer producer = new PeopleEventProducer(ringBuffer);

    Map<String,Object> map = new HashMap<String, Object>();
    map.put("name","yjz");
    map.put("age",25);
    map.put("sex",1);

    producer.onData(map);

}

}

</pre>

5.事件发布

事件发布包括三个步骤：

事件在提交之后才会通知EventProcessor进行处理。

<pre>
long sequence = ringBuffer.next();
try {
PeopleEvent event = ringBuffer.get(sequence);
event.setName(data.get("name").toString());
event.setAge((Integer) data.get("age"));
event.setSex((Integer) data.get("sex"));
}finally {
ringBuffer.publish(sequence);
}
</code></pre>

RingBuffer.publish必须在finally来确保调用，如果某个sequence未被提交，将会阻塞还需发布或其它的producer。
Disruptor提供了另一种方式简化上述操作，来确保publish总被调用：

<pre>
private static final EventTranslatorOneArg<PeopleEvent,Map<String,Object>> tranlator = new EventTranslatorOneArg<PeopleEvent, Map<String, Object>>() {
public void translateTo(PeopleEvent event, long sequence, Map<String, Object> data) {
event.setName(data.get("name").toString());
event.setAge((Integer) data.get("age"));
event.setSex((Integer) data.get("sex"));
}
};

/**
 * onData用来发布事件，每调用一次就发布一次事件，它的参数会通过事件传递给消费者
 */
public void onData(Map<String,Object> data){
    ringBuffer.publishEvent(tranlator,data);
}

</code></pre>

事件发布完整代码

<pre>
/**

6.关闭Disruptor

<pre>
disruptor.shutdown();
executor.shutdown();
</pre>

这里只是Disruptor的一个简介，深入了解后再继续分享。

上一篇下一篇

猜你喜欢

热点阅读