Disruptor的概述與使用

 

一概述

1.Disruptor

Disruptor是一個高效能的非同步處理框架，一個「生產者-消費者」模型。

2.RingBuffer

RingBuffer是一種環形資料結構，包含一個指向下一個槽點的序號，可以在執行緒間傳遞數據。

3.Event

在Disruptor框架中，生產者生產的資料叫做Event。

二 Disruptor框架基本上構成

# 1.MyEvent：自訂對象，充當「生產者-消費者」模型中的資料。
  2.MyEventFactory：實作EventFactory的接口，用於生產資料。
  3.MyEventProducerWithTranslator：將資料儲存到自訂物件中並發布。
  4.MyEventHandler：自訂消費者。

三Demo

 初次接觸Disruptor，認識停留在表面，零散，模糊，在此記一個簡單的示例，以便日後深入研究。

1.自訂資料類別 

package com.disruptor.basic;public class LongEvent {private long value;public long getValue() {return value;
    }public void setValue(long value) {this.value = value;
    }

}

2.資料生產工廠（創建資料類別物件）

package com.disruptor.basic;import com.lmax.disruptor.EventFactory;public class LongEventFactory implements EventFactory<longevent> {public LongEvent newInstance() {// TODO Auto-generated method stubreturn new LongEvent();
    }

}</longevent>

3.資料來源(初始化資料物件並發布)

package com.disruptor.basic;import java.nio.ByteBuffer;import com.lmax.disruptor.EventTranslatorOneArg;import com.lmax.disruptor.RingBuffer;public class LongEventProducerWithTranslator {private final RingBuffer<longevent> ringBuffer;public LongEventProducerWithTranslator(RingBuffer<longevent> ringBuffer) {this.ringBuffer = ringBuffer;
    }private final EventTranslatorOneArg<longevent> TRANSLATOR = new EventTranslatorOneArg<longevent>() {/** * event：包含有消费数据的对象； sequence：分配给目标对象的RingBuffer空间序号；
         * bb：包含有将要被存储到目标对象中的数据的容器         */public void translateTo(LongEvent event, long sequence, ByteBuffer bb) {// TODO Auto-generated method stubevent.setValue(bb.getLong(0));// 将数据存储到目标对象中        }
    };public void onData(ByteBuffer bb) {
        ringBuffer.publishEvent(TRANSLATOR, bb);// 发布，将数据推送给消费者    }

}</longevent></longevent></longevent></longevent>

4.消費者

package com.disruptor.basic;import com.lmax.disruptor.EventHandler;public class LongEventHandler implements EventHandler<longevent> {public void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {// TODO Auto-generated method stubSystem.out.println("当前消费的数据="+event.getValue());
    }

}</longevent>

5.測試類別

package com.disruptor.basic;import java.nio.ByteBuffer;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import org.junit.Test;import com.lmax.disruptor.EventFactory;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.YieldingWaitStrategy;import com.lmax.disruptor.dsl.Disruptor;import com.lmax.disruptor.dsl.ProducerType;public class LongEventTest {

    @SuppressWarnings({ "unchecked", "deprecation" })
    @Testpublic void test01() throws InterruptedException {
        ExecutorService executor = Executors.newCachedThreadPool();
        EventFactory<longevent> factory = new LongEventFactory();int bufferSize = 1024;
        Disruptor<longevent> disruptor = new Disruptor<longevent>(factory, bufferSize, executor, ProducerType.SINGLE,new YieldingWaitStrategy());
        disruptor.handleEventsWith(new LongEventHandler());
        disruptor.start();

        RingBuffer<longevent> ringBuffer = disruptor.getRingBuffer();// LongEventProducer producer = new// LongEventProducer(ringBuffer);LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);
        ByteBuffer bb = ByteBuffer.allocate(8);// long startTime = System.currentTimeMillis();for (long a = 0; a </longevent></longevent></longevent></longevent>

 

以上是Disruptor的概述與使用的詳細內容。更多資訊請關注PHP中文網其他相關文章！