disruptor_harden vol1disruptor_lmax.disruptorr(3)

复杂例子

这个例子中，我们需要有不同的消费者，并且有些消费者之间存在依赖关系，有些消费者之间可以并行处理。disruptor

public static void main(String[] args) throws InterruptedException {
    long beginTime = System.currentTimeMillis();

    int bufferSize = 4;
    ExecutorService executor = Executors.newFixedThreadPool(10);// 大于consumer的数量

    Disruptor<TestObject> disruptor = new Disruptor<TestObject>(new TestObjectFactory(), bufferSize, executor,
            ProducerType.SINGLE, new BusySpinWaitStrategy());

    // //使用disruptor创建消费者AnalysisHandler,CalcHandler，两个可以并行执行
    // EventHandlerGroup<TestObject>
    // handlerGroup=disruptor.handleEventsWith(new
    // TestObjectAnalysisHandler(),new TestObjectCalcHandler());
    //
    // //声明在AnalysisHandler,CalcHandler完事之后执行NotifyHandler
    // EventHandlerGroup<TestObject> then = handlerGroup.then(new
    // TestObjectNotifyHandler());
    //
    // //最终调用多个线程，进行数据的写入
    // then.thenHandleEventsWithWorkerPool(new TestObjectDBHandler(),new
    // TestObjectDBHandler());

    // 上面的也可以直接通过链式调用
    disruptor.handleEventsWith(new TestObjectAnalysisHandler(), new TestObjectCalcHandler())
            .then(new TestObjectNotifyHandler())
            .thenHandleEventsWithWorkerPool(new TestObjectDBHandler(), new TestObjectDBHandler());

    disruptor.start();// 启动

    CountDownLatch latch = new CountDownLatch(1);
    // 生产者准备
    executor.submit(new TestObjectPublisher(latch, disruptor));
    latch.await();// 等待生产者完事.
    disruptor.shutdown();
    executor.shutdown();

    System.out.println("总耗时:" + (System.currentTimeMillis() - beginTime));
}

本文来自电脑杂谈，转载请注明本文网址：
https://www.pc-fly.com/a/jisuanjixue/article-34283-3.html