Wie führt diese Java-App mehrere Threads aus, ohne Threads zu erweitern oder Runnable zu implementieren?

Question

Ich lerne Java und konnte mithilfe von Runnable ein wenig Multithreading mit meinen vorhandenen Apps durchführen.Ich habe mir jetzt Disruptor angesehen (um Variablen zwischen Threads zu teilen), aber ich kann nicht herausfinden, wie der Autor tatsächlich Threads erzeugt.

Ich sehe, dass er Executor verwendet, den ich verwende, um ausführbare Klassen in meinem Programm zu übermitteln, aber in diesem Beispiel gibt es kein Submit (oder ausführbares).Ich habe das, was ich weiß, nur aus den Oracle-Tutorials gelernt, und dort wird erwähnt, dass die einzigen zwei Möglichkeiten darin bestehen, Threads zu erweitern oder Runnable zu implementieren (ich sehe hier keines von beidem, aber er übergibt den Executor an den Disruptor, was könnte sein Threading sein?).Übersehe ich etwas oder macht diese Person es anders?Mein Endziel ist es, diesen Code zu verstehen (der perfekt funktioniert), damit ich ihn auf meinen vorhandenen (ausführbaren) Code anwenden kann.

Hier ist der betreffende Code:

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

Aktualisieren:Wenn Disruptor das Spawnen von Threads übernimmt, wie kann ich dann meine vorhandene ausführbare Klasse an ihn senden?Oder muss ich den Code noch einmal überarbeiten?Tut mir leid, ich bin etwas verwirrt, ob Disruptor mit vorhandenem Code funktionieren wird oder ob ich meine Sachen dafür komplett ändern muss.

Answer 1

Wie Sie vermuten, erfolgt die eigentliche Bearbeitung von Threads (über das Senden von Arbeitselementen) im Inneren Disruptor.Man muss also hinschauen seinen Quellcode (Zu Ihrem Glück ist es Open Source), um Folgendes zu finden:

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