Disruptor使用入门

标签: disruptor | 发表时间:2014-08-04 01:42 | 作者:guimingyue
出处:http://blog.csdn.net

在最近的项目中看到同事使用到了Disruptor,以前在ifeve上看到过关于Disruptor的文章,但是没有深入研究,现在项目中用到了,就借这个机会对这个并发编程框架进行深入学习。项目中使用到的是disruptor-2.10.4,所以下面分析到的Disruptor的代码是这个版本的。

并发编程网介绍Disruptor的文章是disruptor1.0版本,所以有一些术语在2.0版本上已经没有了或者被替代了。

Disruptor术语

github上Disruptor的wiki对Disruptor中的术语进行了解释,在看Disruptor的过程中,对于几个其他的类,觉得有必要与这些术语放到一起,就加进来了。

  • RingBuffer 经常被看作Disruptor最主要的组件,然而从3.0开始RingBuffer仅仅负责存储和更新在Disruptor中流通的数据。对一些特殊的使用场景能够被用户(使用其他数据结构)完全替代。
  • Sequence Disruptor使用Sequence来表示一个特殊组件处理的序号。和Disruptor一样,每个消费者(EventProcessor)都维持着一个Sequence。大部分的并发代码依赖这些Sequence值的运转,因此Sequence支持多种当前AtomicLong类的特性。事实上,这两者之间唯一的区别是Sequence包含额外的功能来阻止Sequence和其他值之间的共享。
  • Sequencer 这是Disruptor真正的核心。实现了这个接口的两种生产者(单生产者和多生产者)均实现了所有的并发算法,为了在生产者和消费者之间进行准确快速的数据传递。
  • SequenceBarrier 由Sequencer生成,并且包含了已经发布的Sequence的引用,这些的Sequence源于Sequencer和一些独立的消费者的Sequence。它包含了决定是否有供消费者来消费的Event的逻辑。
  • WaitStrategy:它决定了一个消费者将如何等待生产者将Event置入Disruptor。
  • Event 从生产者到消费者过程中所处理的数据单元。Disruptor中没有代码表示Event,因为它完全是由用户定义的。
  • EventProcessor 主要的事件循环,用于处理Disruptor中的Event,并且拥有消费者的Sequence。它有一个实现类是BatchEventProcessor,包含了event loop有效的实现,并且将回调到一个EventHandler接口的实现对象。
  • EventHandler 由用户实现并且代表了Disruptor中的一个消费者的接口。
  • Producer 由用户实现,它调用RingBuffer来插入事件(Event),在Disruptor中没有相应的实现代码,由用户实现。
  • WorkProcessor 确保每个sequence只被一个processor消费,在同一个WorkPool中的处理多个WorkProcessor不会消费同样的sequence。
  • WorkerPool 一个WorkProcessor池,其中WorkProcessor将消费Sequence,所以任务可以在实现WorkHandler接口的worker吃间移交
  • LifecycleAware 当BatchEventProcessor启动和停止时,于实现这个接口用于接收通知。

Disruptor印象

初看Disruptor,给人的印象就是RingBuffer是其核心,生产者向RingBuffer中写入元素,消费者从RingBuffer中消费元素,如下图:

这就是Disruptor最简单的模型。其中的RingBuffer被组织成要给环形队列,但它与我们在常常使用的队列又不一样,这个队列大小固定,且每个元素槽都以一个整数进行编号,RingBuffer中只有一个游标维护着一个指向下一个可用位置的序号,生产者每次向RingBuffer中写入一个元素时都需要向RingBuffer申请一个可写入的序列号,如果此时RingBuffer中有可用节点,RingBuffer就向生产者返回这个可用节点的序号,如果没有,那么就等待。同样消费者消费的元素序号也必须是生产者已经写入了的元素序号。

那么Disruptor是如何实现这些逻辑的呢?先来看一个Disruptor的使用示例

Disruptor使用示例

不适用Disruptor的dsl,直接使用Disruptor中的类来完成。

//RingBuffer中存储的单元
public class IntEvent {
    private int value = -1;

    public int getValue() {
        return value;
    }

    public void setValue(int value) {
        this.value = value;
    }

    public String toString() {
        return String.valueOf(value);
    }

    public static EventFactory<IntEvent> INT_ENEVT_FACTORY = new EventFactory<IntEvent>() {
        public IntEvent newInstance() {
            return new IntEvent();
        }
    };
}
//生产者
public class IntEventProducer implements WorkHandler<IntEvent> {

    private int seq = 0;
    public void onEvent(IntEvent event) throws Exception {
        System.out.println("produced " + seq);
        event.setValue(++seq);
    }

}
//消费者
public class IntEventProcessor implements WorkHandler<IntEvent> {

    public void onEvent(IntEvent event) throws Exception {
        System.out.println(event.getValue());
        event.setValue(1);
    }

}

public class DisruptorTest {

    public static void main(String[] args) throws InterruptedException {
        //创建一个RingBuffer对象
        RingBuffer<IntEvent> ringBuffer = new RingBuffer<IntEvent>(IntEvent.INT_ENEVT_FACTORY,
            new SingleThreadedClaimStrategy(16),
            new SleepingWaitStrategy());

        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        IntEventProducer[] producers = new IntEventProducer[1];
        for (int i = 0; i < producers.length; i++) {
            producers[i] = new IntEventProducer();
        }
        WorkerPool<IntEvent> crawler = new WorkerPool<IntEvent>(ringBuffer,
            sequenceBarrier,
            new IntEventExceptionHandler(),
            producers);
        SequenceBarrier sb = ringBuffer.newBarrier(crawler.getWorkerSequences());
        IntEventProcessor[] processors = new IntEventProcessor[1];
        for (int i = 0; i < processors.length; i++) {
            processors[i] = new IntEventProcessor();
        }

        WorkerPool<IntEvent> applier = new WorkerPool<IntEvent>(ringBuffer,sb,
            new IntEventExceptionHandler(),
            processors);
        List<Sequence> gatingSequences = new ArrayList<Sequence>();
        for(Sequence s : crawler.getWorkerSequences()) {
            gatingSequences.add(s);
        }
        for(Sequence s : applier.getWorkerSequences()) {
            gatingSequences.add(s);
        }
ringBuffer.setGatingSequences(gatingSequences.toArray(new Sequence[gatingSequences.size()]));
        ThreadPoolExecutor executor = new ThreadPoolExecutor(7,7,10,TimeUnit.MINUTES,new LinkedBlockingQueue<Runnable>(5));
        crawler.start(executor);
        applier.start(executor);

        while (true) {
            Thread.sleep(1000);
            long lastSeq = ringBuffer.next();
            ringBuffer.publish(lastSeq);
        }
    }
}

class IntEventExceptionHandler implements ExceptionHandler {
    public void handleEventException(Throwable ex, long sequence, Object event) {}
    public void handleOnStartException(Throwable ex) {}
    public void handleOnShutdownException(Throwable ex) {}
}


在上面的代码中IntEvent类就是术语中的Event,IntEventProducer对应Producer,IntEventProcessor对应着EventProcessor,也就是消费者,但是IntEventProcessor类并不是实现的IntEventProcessor接口,这个下面会分析到。

下面从main方法开始分析。

首先是RingBuffer的创建, RingBuffer<IntEvent> ringBuffer = new RingBuffer<IntEvent>(IntEvent.INT_ENEVT_FACTORY,new SingleThreadedClaimStrategy(16), new SleepingWaitStrategy());RingBuffer的构造方法的第一个参数是实现了EventFactory接口的类,主要作用是创建IntEvent对象,在创建RingBuffer对象时,第二个参数是 ClaimStrategy,生产者通过 ClaimStrategy 来申请下一个可用节点,第三个参数是 WaitStrategy的实现类,它定义了消费者的等待策略。

下面一段代码是初始化生产者的过程

SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
IntEventProducer[] producers = new IntEventProducer[1];
for (int i = 0; i < producers.length; i++) {
    producers[i] = new IntEventProducer();
}
WorkerPool<IntEvent> crawler = new WorkerPool<IntEvent>(ringBuffer,
sequenceBarrier,new IntEventExceptionHandler(), producers);

ringBuffer.newBarrier()返回的是一个SequenceBarrier对象,Barrier顾名思义就是障碍的意思,这个障碍阻止了生产者越过其所能到达的ringbuffer中的位置。RingBuffer在执行过程中会维持一个游标cursor,所以生产者从RingBuffer中获取到的的游标必须小于等于这个cursor。

然后就是消费者的初始化:

SequenceBarrier sb = ringBuffer.newBarrier(crawler.getWorkerSequences());
IntEventProcessor[] processors = new IntEventProcessor[1];
for (int i = 0; i < processors.length; i++) {
    processors[i] = new IntEventProcessor();
}

WorkerPool<IntEvent> applier = new WorkerPool<IntEvent>(ringBuffer,sb,new IntEventExceptionHandler(),processors);

消费者初始化也需要设置一个SequenceBarrier对象,这个SequenceBarrier对象指明了消费者可以消费的元素序号,如果消费者的游标大于这个序号,那么消费者必须以 WaitStrategy定义的策略等待。

生产者和消费者创建完成后下一步是设置 RingBuffer的一个变量 gatingSequencesgatingSequences的作用是防止生产者覆盖还未被消费者消费的元素,假设一个RingBuffer的大小为8,消费者消费速度较慢,那么RingBuffer可能是满的,当生产者向RingBuffer申请下一个可用序号时,还未被消费者消费的序号就不能被覆盖,所以RingBuffer就不能给生产者返回可用序号,此时消费者线程就进入等待,在这种情况下RingBuffer会检查当前申请的序号是否大于 gatingSequences中的最小序号,如果当前申请的序号大于最小序号,那么生产者就等待。代码如下:

List<Sequence> gatingSequences = new ArrayList<Sequence>();
for(Sequence s : crawler.getWorkerSequences()) {
    gatingSequences.add(s);
}
for(Sequence s : applier.getWorkerSequences()) {
    gatingSequences.add(s);
}     ringBuffer.setGatingSequences(gatingSequences.toArray(new Sequence[gatingSequences.size()]));


这里之所以加入了生产者的序号,是因为在多生产者的情况下,一个生产者不能覆盖另一个生产者已经申请的序号。

然后下面一段的代码就是启动Disruptor线程来执行生产者和消费者线程。

ThreadPoolExecutor executor = new ThreadPoolExecutor(7, 7,10,
TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(2));
crawler.start(executor);
 applier.start(executor);

最后要做的是生产者不断的申请可用序号,提交准备好的元素序号,即:

while (true) {
    Thread.sleep(1000);
    long lastSeq = ringBuffer.next();
    ringBuffer.publish(lastSeq);
}


这个过程一般成为两阶段提交,第一个阶段调用  RingBuffer.next()方法得到下一个可用节点,此时会调用生产者类的OnEvent方法,即  IntEventProducer.onEvent(),第二个阶段是发布已经生产完成的元素序号  lastSeq,提醒消费者,可以消费这个序号了。

小结

本文简要介绍了Disruptor的一些术语,解释了一个Disruptor示例程序执行过程的各个要点,从宏观上了解了Disruptor的执行流程,在下一篇文章中接着分析Disruptor的内部执行原理。

Reference

并发框架Disruptor译文
JAVA并发编程学习笔记之Unsafe类
从 Java 代码到 Java 堆
Disruptor的GitHub
Disruptor使用指南

作者:guimingyue 发表于2014-8-3 17:42:41 原文链接
阅读:61 评论:0 查看评论

相关 [disruptor] 推荐:

Disruptor封装

- - 开源软件 - ITeye博客
在数据交换场景,disruptor受到越来越多的欢迎. 下面是将原生disruptor封装成queue模型的代码,供参考. 抽象类Disruptor,提供pull、take等接口. 已有 0 人发表留言,猛击->> 这里<<-参与讨论. —软件人才免语言低担保 赴美带薪读研.

disruptor使用示例

- - 企业架构 - ITeye博客
LMAX 开源了一个高性能并发编程框架. 可以理解为消费者-生产者的消息发布订阅模式. 本文下载了官方示例代码,进行实验. longEvent事件数据. LongEventFactory事件工厂. import com.lmax.disruptor.EventFactory; /** * 事件生产工厂 * @author wanghao * */ public class LongEventFactory implements EventFactory {.

Disruptor 学习笔记

- - 开源软件 - ITeye博客
Disruptor 是一个高性能异步处理框架,也可以认为是一个消息框架,它实现了观察者模式. Disruptor 比传统的基于锁的消息框架的优势在于:它是无锁的、CPU友好;它不会清除缓存中的数据,只会覆盖,降低了垃圾回收机制启动的频率. Disruptor 为什么快. 通过内存屏障和原子性的CAS操作替换锁.

Disruptor使用入门

- - CSDN博客推荐文章
在最近的项目中看到同事使用到了Disruptor,以前在ifeve上看到过关于Disruptor的文章,但是没有深入研究,现在项目中用到了,就借这个机会对这个并发编程框架进行深入学习. 项目中使用到的是disruptor-2.10.4,所以下面分析到的Disruptor的代码是这个版本的. 并发编程网介绍Disruptor的文章是disruptor1.0版本,所以有一些术语在2.0版本上已经没有了或者被替代了.

并发框架Disruptor译文

- - 酷壳 - CoolShell.cn
(感谢同事 方腾飞投递本文). Martin Fowler在自己网站上写了一篇LMAX架构的文章,在文章中他介绍了LMAX是一种新型零售金融交易平台,它能够以很低的延迟产生大量交易. 这个系统是建立在JVM平台上,其核心是一个业务逻辑处理器,它能够在一个线程里每秒处理6百万订单. 业务逻辑处理器完全是运行在内存中,使用事件源驱动方式.

Disruptor 极速体验 - haiq

- - 博客园_首页
      已经不记得最早接触到 Disruptor 是什么时候了,只记得发现它的时候它是以具有闪电般的速度被介绍的. 于是在脑子里, Disruptor 和“闪电”一词关联了起来,然而却一直没有时间去探究一下.       最近正在进行一项对性能有很高要求的产品项目的研究,自然想起了闪电般的 Disruptor ,这必有它的用武之地,于是进行了一番探查,将成果和体会记录在案.

Disruptor为何这么快

- - 掘金后端
Disruptor 是一个开源并且高效的 生产者-消费者框架,很难直接解释这个框架是做什么的,但是可以把这个框架理解成 Java 中的 BlockingQueue. 这样理解起来是不是轻松多了,这就是一个生产者-消费者队列,只不过它的性能要比 BloockingQueue 好很多,号称单机器可以有百万的 TPS.

GitHub - chanjarster/artemis-disruptor-miaosha: 没有redis也能够支撑

- -
"小米在印度把亚马逊搞挂了"事件的秒杀解决方案. 小米在印度打破了多项记录:. 4分钟内卖出了超过250,000台. ---OPS:1042次抢购/S. 抢购前我们收到了100万“到货提醒”. 亚马逊每分钟收到超过500万个点击. 亚马逊在这个期间每秒收到1500个订单(这是印度电商公司所有销售中最高的).

还在用阻塞队列?读这篇文章,了解下 Disruptor 吧

- - IT瘾-dev
听到队列相信大家对其并不陌生,在我们现实生活中队列随处可见,去超市结账,你会看见大家都会一排排的站得好好的,等待结账,为什么要站得一排排的,你想象一下大家都没有素质,一窝蜂的上去结账,不仅让这个超市崩溃,还会容易造成各种踩踏事件,当然这些事其实在我们现实中也是会经常发生. 当然在计算机世界中,队列是属于一种数据结构,队列采用的FIFO(first in firstout),新元素(等待进入队列的元素)总是被插入到尾部,而读取的时候总是从头部开始读取.