이 Java 응용 프로그램은 스레드를 확장하거나 실행 가능을 구현하지 않고 여러 스레드를 어떻게 실행합니까?

Question

저는 Java를 배우고 있으며 기존 응용 프로그램에서 runnable을 사용하여 멀티 스레딩을 할 수있었습니다. 나는 현재 스레드 (thread간에 변수를 공유하기 위해)를 혼란시키는 사람을보고 있었지만, 저자가 실제로 스레드를 생성하는 방법을 알 수는 없다.

나는 그가 내 프로그램에서 실행 가능한 클래스를 제출하는 데 사용하는 Executor를 사용하고 있지만이 예제에서는 제출 (또는 실행)이 없다고 본다. 나는 오라클 튜토리얼에서 알 수있는 것을 배웠고 스레드를 확장하거나 실행 가능한 도구를 구현하는 것이 두 가지 방법 밖에 없다고 언급했다. (나는 여기도 보지 못하지만 집행자를 방해자에게 제출한다. 나는 뭔가를 놓치고 있는가? 아니면이 사람이 다른 방식으로 그것을하고 있는가? 내 최종 목표는이 코드를 이해하는 것입니다 (완벽하게 작동합니다). 그래서 그것을 기존의 (실행 가능을 사용하여) 코드에 적용 할 수 있습니다.

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

업데이트 :

여기에 문제의 코드는 스럽 그때 내가 제출하는 방법을 내 그것에 실행 가능한 클래스를 기존 스레드의 산란을 처리하는 경우? 또는 코드를 다시 수정해야합니까? 죄송합니다. 혼란스러워하는 개발자가 기존 코드를 사용하거나 완전히 수정해야 할 경우 혼란 스럽습니다.

Answer 1

실제 스레드 처리 (작업 항목 제출을 통한)는 Disruptor 내부에서 수행됩니다. 그래서 만약 당신이 (당신의 행운에, 그것은 오픈 소스) its source code 볼 필요가,이를 찾을 수 :

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