快捷導(dǎo)航

Java多線程之Disruptor入門

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

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

一、Disruptor簡介

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

二、淺聊Disruptor的核心

在這里插入圖片描述

Disruptor維護(hù)了一個(gè)環(huán)形隊(duì)列RingBuffer，這個(gè)隊(duì)列本質(zhì)上是一個(gè)首位相連的數(shù)組。相比于LinkedBlockdingQueue，RingBuffer的數(shù)組結(jié)構(gòu)在查找方面效率更高。此外，LinkedBlockingQueue需要維護(hù)一個(gè)頭節(jié)點(diǎn)指針head和一個(gè)尾節(jié)點(diǎn)指針tail，而RingBuffer只需要維護(hù)一個(gè)sequence指向下一個(gè)可用的位置即可。所以從這兩點(diǎn)來說，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)者消費(fèi)者模型。其RingBuffer中存放的其實(shí)是將消息封裝成的事件。這里定義了一個(gè)LongEvent，表示消息隊(duì)列中存放的是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

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

import com.lmax.disruptor.EventFactory;

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

3.4 EventHandler

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

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;

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

    private RingBuffer<LongEvent> ringBuffer;

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

    public void onData(ByteBuffer byteBuffer) {
        // 定位到下一個(gè)可存放的位置
        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)造一個(gè)Disruptor
        Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, Executors.defaultThreadFactory());
        // 綁定handler
        disruptor.handleEventsWith(new LongEventHandler());

        // 啟動(dòng)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)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: