2

我正在学习 Java，并且能够使用 runnable 对我现有的应用程序进行一些多线程处理。我现在正在查看中断器（在线程之间共享变量），但我无法弄清楚作者实际上是如何产生线程的。

我看到他正在使用 Executor，我用它在我的程序中提交可运行的类，但在这个例子中没有提交（或可运行）。我只从 Oracle 教程中学到了我所知道的知识，他们提到唯一的两种方法是扩展线程或实现可运行（我在这里都没有看到，但他确实将执行程序提交给了中断器，这可能是他如何处理线程？）。我错过了什么还是这个人以不同的方式做这件事？我的最终目标是理解这段代码（它工作得很好），所以我可以将它应用到我现有的（使用可运行的）代码中。

这是有问题的代码：

应用程序.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);
    }
}

更新：如果 Disruptor 正在处理线程的产生，那么我如何将现有的可运行类提交给它？还是我需要再次修改代码？抱歉，我对中断器是否要使用现有代码或者我是否需要完全改变我的东西感到有点困惑。

4

1 回答 1

7

正如您所怀疑的，实际处理线程（通过提交工作项）是在Disruptor. 所以你需要查看它的源代码（幸运的是，它是开源的），找到这个：

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;
}
于 2012-03-13T17:20:58.687 回答