Disruptor PK BlockingQueue

 

package com.disruptor.test3;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;

import org.junit.Test;

public class ArrayBlockingQueueTest {

    @Test
    public void test() throws InterruptedException {

        long cost = System.currentTimeMillis();

        final CountDownLatch l = new CountDownLatch(1);
        final BlockingQueue<Long> bq = new ArrayBlockingQueue<Long>(4096);

        Runnable p = new Runnable() {
            public void run() {
                for (int i = 0; i < ConstantsUtil.MAX_LOOP; i++) {
                    try {
                        bq.put((long) i);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };

        Runnable c = new Runnable() {
            public void run() {
                while (true) {
                    try {
                        long i = bq.take();
                        //System.out.println(i);
                        if (i == ConstantsUtil.MAX_LOOP - 1) {
                            l.countDown();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };

        new Thread(c).start();

        new Thread(p).start();

        l.await();
        System.out.println("cost:" + (System.currentTimeMillis() - cost));
    }

}

 

 

package com.disruptor.test3;

public abstract class ConstantsUtil {
    static int MAX_LOOP = 1000000000;
}

 

 

package com.disruptor.test3;

import java.util.concurrent.CountDownLatch;

import org.junit.Test;

import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.YieldingWaitStrategy;

public class DisruptorTest {

    private static final int BUFFER_SIZE = 4096;

    @Test
    public void test() throws InterruptedException {
        long cost = System.currentTimeMillis();
        CountDownLatch l = new CountDownLatch(1);
        //创建RingBuffer
        RingBuffer<ValueEvent> ringBuffer = RingBuffer.createSingleProducer(
            ValueEvent.EVENT_FACTORY, BUFFER_SIZE, new YieldingWaitStrategy());

        //创建序列栅栏
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        //创建消费者
        MyEventHandler handler = new MyEventHandler(l);

        //事件执行者
        BatchEventProcessor<ValueEvent> batchEventProcessor = new BatchEventProcessor<ValueEvent>(
            ringBuffer, sequenceBarrier, handler);

        //序列由栅栏统一计算
        ringBuffer.addGatingSequences(batchEventProcessor.getSequence());

        new Thread(batchEventProcessor).start();

        for (long i = 0; i < ConstantsUtil.MAX_LOOP; i++) {
            long next = ringBuffer.next();
            //通过序列从环中，获取消息,没有则由ValueEvent.EVENT_FACTORY工厂创建空事件
            ValueEvent event = ringBuffer.get(next);
            //填充数据
            event.setValue(i);
            //将环中的数据发布出去，发布之后，实际也是直接通过事件消费
            ringBuffer.publish(next);
        }
        l.await();
        System.out.println("cost:" + (System.currentTimeMillis() - cost));
    }
}

 

 

package com.disruptor.test3;

import java.util.concurrent.CountDownLatch;

import com.lmax.disruptor.EventHandler;

public class MyEventHandler implements EventHandler<ValueEvent> {
    private CountDownLatch l;

    public long            count = 0;

    public MyEventHandler() {
    };

    public MyEventHandler(CountDownLatch l) {
        this.l = l;
    };

    public void onEvent(ValueEvent event, long arg1, boolean arg2) throws Exception {
        long i = event.getValue();
        //System.out.println(i);
        if (i == ConstantsUtil.MAX_LOOP - 1) {
            l.countDown();
        }
    }
}

 

 

package com.disruptor.test3;

import com.lmax.disruptor.EventFactory;

public final class ValueEvent {
    private long value;

    public long getValue() {
        return value;
    }

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

    public final static EventFactory<ValueEvent> EVENT_FACTORY = new EventFactory<ValueEvent>() {
                                                                   public ValueEvent newInstance() {
                                                                       return new ValueEvent();
                                                                   }
                                                               };
}

 

 

1亿：
Disrupter :3910 ms       
BlockQueue: 246211 ms

4s VS 25s   6倍

10亿
Disrupter :36767 ms       
BlockQueue: 231872 ms

37s vs 232s 6倍