Überblick und Verwendung von Disruptor

Ein Überblick

1. Disruptor

Disruptor ist ein Hochleistungsfähiges asynchrones Verarbeitungsframework, ein „Produzenten-Konsumenten“-Modell.

2. RingBuffer

RingBuffer ist eine Ringdatenstruktur, die eine Sequenznummer enthält, die auf den nächsten Slot zeigt und zwischen Threads Daten übergeben werden kann .

3.Ereignis

Im Disruptor-Framework werden die vom Produzenten erzeugten Daten als Ereignis bezeichnet.

Zwei grundlegende Komponenten des Disruptor-Frameworks

1.MyEvent: Benutzerdefiniertes Objekt, das als Daten im „Produzenten-Konsumenten“-Modell fungiert.
2.MyEventFactory: Implementiert die Schnittstelle von EventFactory und wird zum Erzeugen von Daten verwendet.
3.MyEventProducerWithTranslator: Daten in einem benutzerdefinierten Objekt speichern und veröffentlichen.
4.MyEventHandler: Benutzerdefinierter Verbraucher.

Drei Demos

Als ich zum ersten Mal mit Disruptor in Kontakt kam, war mein Verständnis oberflächlich, verstreut und vage. Ich werde eine einfache aufschreiben Beispiel hier, damit ich in Zukunft tiefer in die Forschung einsteigen kann.

1. Angepasste Datenklasse

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

}

2. Datenklassenobjekte erstellen)

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();
    }

}

3. Datenquelle (Datenobjekte initialisieren und veröffentlichen)

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, ByteBuffer> TRANSLATOR = new EventTranslatorOneArg<LongEvent, ByteBuffer>() {/** * 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);// 发布，将数据推送给消费者    }

}

4. Verbraucher

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());
    }

}

5. Testklasse

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 < 100; a++) {
            bb.putLong(0, a);
            producer.onData(bb);/*if (a == 99) {
                long endTime = System.currentTimeMillis();
                System.out.println("useTime=" + (endTime - startTime));
            }*/Thread.sleep(100);
        }/*long endTime = System.currentTimeMillis();
        System.out.println("useTime=" + (endTime - startTime));*/disruptor.shutdown();
        executor.shutdown();
    }/*@Test
    public void test02() {
        long startTime = System.currentTimeMillis();
        for (long a = 0; a < 100; a++) {
            System.out.println(a);
        }
        long endTime = System.currentTimeMillis();
        System.out.println("useTime=" + (endTime - startTime));
    }*/}

Das obige ist der detaillierte Inhalt vonÜberblick und Verwendung von Disruptor. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!