Présentation et utilisation de Disruptor

Un aperçu

1. Le perturbateur

Le perturbateur est un Framework de traitement asynchrone haute performance, un modèle « producteur-consommateur ».

2. RingBuffer

RingBuffer est une structure de données en anneau qui contient un numéro de séquence pointant vers l'emplacement suivant et peut être transmise entre les données des threads .

3.Événement

Dans le framework Disruptor, les données produites par le producteur sont appelées Événement.

Deux composants de base du framework Disruptor

1.MyEvent : Objet personnalisé, agissant comme des données dans le modèle « producteur-consommateur ».
2.MyEventFactory : Implémente l'interface d'EventFactory et est utilisé pour produire des données.
3.MyEventProducerWithTranslator : stockez les données dans un objet personnalisé et publiez-le.
4.MyEventHandler : consommateur personnalisé.

Trois démos

Lorsque je suis entré en contact pour la première fois avec Disruptor, ma compréhension était superficielle, dispersée et vague. Je vais en écrire une simple. exemple ici afin que je puisse approfondir la recherche à l'avenir.

1. Classe de données personnalisées

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

}

2. Créer des objets de classe de données)

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. Source de données (initialiser les objets de données et publier)

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. Consommateur

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. >

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));
    }*/}

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!