快捷導航

Java多線程之Disruptor入門

更新時間：2021年04月29日 09:31:23 作者：EileenChang

這篇文章主要介紹了Java多線程之Disruptor入門,文中有非常詳細的代碼示例,對正在學習java多線程的小伙伴們有非常好的幫助,需要的朋友可以參考下

一、Disruptor簡介

Disruptor目前是世界上最快的單機消息隊列，由英國外匯交易公司LMAX開發(fā)，研發(fā)的初衷是解決內(nèi)存隊列的延遲問題（在性能測試中發(fā)現(xiàn)竟然與I/O操作處于同樣的數(shù)量級）?；贒isruptor開發(fā)的系統(tǒng)單線程能支撐每秒600萬訂單，2010年在QCon演講后，獲得了業(yè)界關(guān)注。2011年，企業(yè)應(yīng)用軟件專家Martin Fowler專門撰寫長文介紹。同年它還獲得了Oracle官方的Duke大獎。目前，包括Apache Storm、Camel、Log4j 2在內(nèi)的很多知名項目都應(yīng)用了Disruptor以獲取高性能。

二、淺聊Disruptor的核心

在這里插入圖片描述

Disruptor維護了一個環(huán)形隊列RingBuffer，這個隊列本質(zhì)上是一個首位相連的數(shù)組。相比于LinkedBlockdingQueue，RingBuffer的數(shù)組結(jié)構(gòu)在查找方面效率更高。此外，LinkedBlockingQueue需要維護一個頭節(jié)點指針head和一個尾節(jié)點指針tail，而RingBuffer只需要維護一個sequence指向下一個可用的位置即可。所以從這兩點來說，RingBuffer比LinkedBlockingQueue要快。

三、Disruptor使用

3.1 pom.xml

<dependency>
            <groupId>com.lmax</groupId>
            <artifactId>disruptor</artifactId>
            <version>3.4.3</version>
        </dependency>

3.2 事件Event

Disruptor是基于事件的生產(chǎn)者消費者模型。其RingBuffer中存放的其實是將消息封裝成的事件。這里定義了一個LongEvent，表示消息隊列中存放的是long類型的數(shù)據(jù)。

public class LongEvent {
	private long value;

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

    @Override
    public String toString() {
        return "LongEvent{" +
                "value=" + value +
                '}';
    }
}

3.3 EventFactory

實現(xiàn)EventFactory接口，定義Event工廠，用于填充隊列。Event工廠其實是為了提高Disruptor的效率，初始化的時候，會調(diào)用Event工廠，對RingBuffer進行內(nèi)存的提前分配，GC的頻率會降低。

import com.lmax.disruptor.EventFactory;

public class LongEventFactory implements EventFactory<LongEvent> {
	public LongEvent newInstance() {
		return new LongEvent();
	}
}

3.4 EventHandler

實現(xiàn)EventHandler接口，定義EventHandler(消費者)，處理容器中的元素。

import com.lmax.disruptor.EventHandler;

public class LongEventHandler implements EventHandler<LongEvent> {
	public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {
		System.out.println("Event: " + event + ", sequence: " + sequence);
	}
}

3.5 使用Disruptor原始API發(fā)布消息

import cn.flying.space.disruptor.demo.LongEvent;
import com.lmax.disruptor.RingBuffer;

import java.nio.ByteBuffer;

/**
 * 定義一個生產(chǎn)者，往Disruptor中投遞消息
 */
public class LongEventProducer {

    private RingBuffer<LongEvent> ringBuffer;

    public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    public void onData(ByteBuffer byteBuffer) {
        // 定位到下一個可存放的位置
        long sequence = ringBuffer.next();
        try {
            // 拿到該位置的event
            LongEvent event = ringBuffer.get(sequence);
            // 設(shè)置event的值
            event.set(byteBuffer.getLong(0));
        } finally {
            // 發(fā)布
            ringBuffer.publish(sequence);
        }
    }
}

import cn.flying.space.disruptor.demo.LongEvent;
import cn.flying.space.disruptor.demo.LongEventFactory;
import cn.flying.space.disruptor.demo.LongEventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

import java.nio.ByteBuffer;
import java.util.concurrent.Executors;
public class TestMain {
    public static void main(String[] args) throws InterruptedException {
        // 定義event工廠
        LongEventFactory factory = new LongEventFactory();
        // ringBuffer長度
        int bufferSize = 1024;
        // 構(gòu)造一個Disruptor
        Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, Executors.defaultThreadFactory());
        // 綁定handler
        disruptor.handleEventsWith(new LongEventHandler());

        // 啟動Disruptor
        disruptor.start();
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        LongEventProducer producer = new LongEventProducer(ringBuffer);

        ByteBuffer byteBuffer = ByteBuffer.allocate(8);
        for (long i = 0; true; i++) {
            byteBuffer.clear();
            byteBuffer.putLong(i);
            // 投遞消息
            producer.onData(byteBuffer);
            Thread.sleep(1000);
        }
    }
}

3.6 使用Translators發(fā)布消息

import cn.flying.space.disruptor.demo.LongEvent;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;

import java.nio.ByteBuffer;

public class LongEventProducerUsingTranslator {
    private RingBuffer<LongEvent> ringBuffer;
    public LongEventProducerUsingTranslator(RingBuffer<LongEvent> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }

    private static final EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR = new EventTranslatorOneArg<LongEvent, ByteBuffer>() {
        @Override
        public void translateTo(LongEvent longEvent, long l, ByteBuffer byteBuffer) {
            longEvent.set(byteBuffer.getLong(0));
        }
    };

    public void onData(ByteBuffer byteBuffer) {
        ringBuffer.publishEvent(TRANSLATOR, byteBuffer);
    }
}

import cn.flying.space.disruptor.demo.LongEvent;
import cn.flying.space.disruptor.demo.LongEventFactory;
import cn.flying.space.disruptor.demo.LongEventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;

import java.nio.ByteBuffer;

/**
 * @author ZhangSheng
 * @date 2021-4-26 14:23
 */
public class TestMain {

    public static void main(String[] args) throws InterruptedException {
        LongEventFactory factory = new LongEventFactory();
        int bufferSize = 1024;
        Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, DaemonThreadFactory.INSTANCE);
        disruptor.handleEventsWith(new LongEventHandler());

        disruptor.start();
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        LongEventProducerUsingTranslator producer = new LongEventProducerUsingTranslator(ringBuffer);
        ByteBuffer byteBuffer = ByteBuffer.allocate(8);

        for (long i = 0L; true; i++) {
            byteBuffer.putLong(0, i);
            // 發(fā)布
            producer.onData(byteBuffer);
            Thread.sleep(1000);
        }
    }
}

到此這篇關(guān)于Java多線程之Disruptor入門的文章就介紹到這了,更多相關(guān)Java Disruptor入門內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: