disruptor 为什么快_disruptor_disrupter(4)

b)异步-事件驱动：经过a)的修改，如果还存在一些业务逻辑处理过程是需要长时间才能完成的，那么就把它作为一个事件，再抛给其他组件（可能还是Disruptor）等待。disruptor业务逻辑处理器需要时刻保持最快速度、最高效率，它不能等待任何事情。

c)每个业务逻辑处理器是单线程的：你没有听错。其实有了a)b)作为前提，会发现多线程所带来的业务层面同步问题将会极大限制BLP效率、增大BLP的复杂度，和BLP的设计（Keep it , stupid.）相悖，如果实在想多线程，可以参照d)。

d)使用多级业务逻辑处理器：有些像管道模式，上图的3块结构可以以多种方式组合，一个BLP可以将输出送往多个Output Disruptor，而这些Disruptor可能是另一些3块结构的InputDisruptor，即有些BLP是起到分发作用的，另一些是进行具体业务逻辑计算的。每个BLP对应一个线程，整个架构可能比上图复杂很多。

3、Hello Disruptor

Disruptor最初是由Java实现的，现在也有C/Cpp和.Net版本，Java版最全更新最快，代码注释较多比较好懂。说了这么多，本节先给出一个测试例子，展示Disruptor的基本用法，例子中用LinkedBlockingQueue和Disruptor分别实现了单一生产者+单一消费者存取简单对象的测试，统计了一下双方消耗的时间，仅供参考。

例子中使用Disruptor 3.2.1。不同版本间的Disruptor一些术语可能有变化，在该版本中，缓冲区里的元素被称作Event，指针（缓冲区的下标）被称作Sequence，生产者的指针为RingBuffer.sequencer（private成员），消费者的指针通过ringBufferInstance.newBarrier()得到。

//简单对象：缓冲区中的元素，里面只有一个value，提供setValueprivate class TestObj {        public long value;        public TestObj(long value)    {        this.value = value;    }        public void setValue(long value)    {        this.value = value;    }    }public class Test {    //待生产的对象个数    final long objCount = 1000000;    final long bufSize;//缓冲区大小    {        bufSize = getRingBufferSize(objCount);    }        //获取RingBuffer的缓冲区大小（2的幂次！加速计算）    static long getRingBufferSize(long num)    {        long s = 2;        while ( s < num )        {            s <<= 1;        }        return s;    }        //使用LinkedBlockingQueue测试    public void testBlocingQueue() throws Exception    {        final LinkedBlockingQueue<TestObj> queue = new LinkedBlockingQueue<TestObj>();        Thread producer = new Thread(new Runnable() {//生产者            @Override            public void run() {                try{                    for ( long i=1;i<=objCount;i++ )                    {                        queue.put(new TestObj(i));//生产                    }                }catch ( InterruptedException e ){                }            }        });        Thread consumer = new Thread(new Runnable() {//消费者            @Override            public void run() {                try{                    TestObj readObj = null;                    for ( long i=1;i<=objCount;i++ )                    {                        readObj = queue.take();//消费                        //DoSomethingAbout(readObj);                    }                }catch ( InterruptedException e ){                }            }        });                long timeStart = System.currentTimeMillis();//统计时间        producer.start();        consumer.start();        consumer.join();        producer.join();        long timeEnd = System.currentTimeMillis();        DecimalFormat df = (DecimalFormat) DecimalFormat.getInstance();        System.out.println((timeEnd - timeStart) + "/" + df.format(objCount) +                " = " + df.format(objCount/(timeEnd - timeStart)*1000) );    }        //使用RingBuffer测试    public void testRingBuffer() throws Exception    {        //创建一个单生产者的RingBuffer，EventFactory是填充缓冲区的对象工厂        //            YieldingWaitStrategy等"等待策略"指出消费者等待数据变得可用前的策略        final RingBuffer<TestObj> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<TestObj>() {            @Override            public TestObj newInstance() {                return new TestObj(0);            }        } , (int)bufSize, new YieldingWaitStrategy());        //创建消费者指针        final SequenceBarrier barrier = ringBuffer.newBarrier();                Thread producer = new Thread(new Runnable() {//生产者            @Override            public void run() {                for ( long i=1;i<=objCount;i++ )                {                    long index = ringBuffer.next();//申请下一个缓冲区Slot                    ringBuffer.get(index).setValue(i);//对申请到的Slot赋值                    ringBuffer.publish(index);//发布，然后消费者可以读到                }            }        });        Thread consumer = new Thread(new Runnable() {//消费者            @Override            public void run() {                TestObj readObj = null;                int readCount = 0;                long readIndex = Sequencer.INITIAL_CURSOR_VALUE;                while ( readCount < objCount )//读取objCount个元素后结束                {                    try{                        long nextIndex = readIndex + 1;//当前读取到的指针+1，即下一个该读的位置                        long availableIndex = barrier.waitFor(nextIndex);//等待直到上面的位置可读取                        while ( nextIndex <= availableIndex )//从下一个可读位置到目前能读到的位置(Batch!)                        {                            readObj = ringBuffer.get(nextIndex);//获得Buffer中的对象                            //DoSomethingAbout(readObj);                            readCount++;                            nextIndex ++;                        }                        readIndex = availableIndex;//刷新当前读取到的位置                    }catch ( Exception ex)                    {                        ex.printStackTrace();                    }                }            }        });                long timeStart = System.currentTimeMillis();//统计时间        producer.start();        consumer.start();        consumer.join();        producer.join();        long timeEnd = System.currentTimeMillis();        DecimalFormat df = (DecimalFormat) DecimalFormat.getInstance();        System.out.println((timeEnd - timeStart) + "/" + df.format(objCount) +                " = " + df.format(objCount/(timeEnd - timeStart)*1000) );            }        public static void main(String[] args) throws Exception {        Test ins = new Test();        //执行测试        ins.testBlocingQueue();        ins.testRingBuffer();    }}

本文来自电脑杂谈，转载请注明本文网址：
http://www.pc-fly.com/a/jisuanjixue/article-26632-4.html