Comment est-ce java application à l'exécution de plusieurs threads sans l'extension de threads ou de la mise en œuvre de praticable?

Question

Je suis en train d'apprendre Java et a été en mesure de faire un peu de multi-threading avec mes applications existantes à l'aide de praticable.J'étais maintenant à la recherche à perturbateur(pour partager des variables entre les threads), mais je ne peux pas comprendre comment l'auteur est en fait la ponte des threads.

Je vois qu'il est à l'aide de l'Exécuteur, que j'utilise pour présenter praticable classes dans mon programme mais dans cet exemple, il n'y a pas de présenter(ou exécutable).Je n'ai appris ce que je sais de l'Oracle tutoriels et ils mentionnent les deux seules façons d'étendre les threads ou de mettre en œuvre runnable(je ne vois pas que ce soit ici, mais il ne soumettre exécuteur de perturbateur, qui peut-être comment il est threading?).Ai-je raté quelque chose ou cette personne est-elle le faire d'une manière différente?Mon objectif final est de comprendre ce code(qui fonctionne parfaitement) pour que je puisse l'appliquer à mon existantes (à l'aide d'exécutables) du code.

Voici le code en question:

App.java

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.*;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

public class App {

    private final static int RING_SIZE = 1024 * 8;

    private static long handleCount = 0;

    private final static long ITERATIONS = 1000L * 1000L * 300L;
    private final static int NUM_EVENT_PROCESSORS = 8;

    private final static EventHandler<ValueEvent> handler =
        new EventHandler<ValueEvent>() {
        public void onEvent(final ValueEvent event,
                                final long sequence,
                                final boolean endOfBatch) throws Exception {
        handleCount++;
    }
    };

    public static void main(String[] args) {
    System.out.println("Starting disruptor app.");

    ExecutorService executor = Executors.newFixedThreadPool(NUM_EVENT_PROCESSORS);

    Disruptor<ValueEvent> disruptor =
        new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, executor,
            new SingleThreadedClaimStrategy(RING_SIZE),
            new SleepingWaitStrategy());
    disruptor.handleEventsWith(handler);
    RingBuffer<ValueEvent> ringBuffer = disruptor.start();

    long start = System.currentTimeMillis();

        long sequence;
        ValueEvent event;
    for (long x=0; x<ITERATIONS; x++) {
        sequence = ringBuffer.next();
        event = ringBuffer.get(sequence);
        event.setValue(x);
        ringBuffer.publish(sequence);
        }
    final long expectedSequence = ringBuffer.getCursor();

    while (handleCount < expectedSequence) { }

    long opsPerSecond = (ITERATIONS * 1000L) / (System.currentTimeMillis() - start);
    System.out.printf("op/s: %d, handled: %d", opsPerSecond, handleCount);
    }
}

Mise à jour:si Perturbateur est de la manipulation de la reproduction de threads alors comment puis-je soumettre mon classe runnable à elle?ou ai-je besoin de retravailler à nouveau le code?Désolé, je suis un peu confus sur si perturbateur va travailler avec du code existant ou si j'ai besoin de changer complètement mes trucs pour elle.

Answer 1

Comme vous le soupçonnez, la négociation avec les threads (par l'intermédiaire de soumettre des éléments de travail) est fait à l'intérieur Disruptor.Si vous avez besoin de regarder son code source (pour votre chance, c'est de l'open source), pour trouver ceci:

public RingBuffer<T> start()
{
    EventProcessor[] gatingProcessors = eventProcessorRepository.getLastEventProcessorsInChain();
    ringBuffer.setGatingSequences(Util.getSequencesFor(gatingProcessors));

    checkOnlyStartedOnce();
    for (EventProcessorInfo<T> eventProcessorInfo : eventProcessorRepository)
    {
        executor.execute(eventProcessorInfo.getEventProcessor());
    }

    return ringBuffer;
}