public class DTaskFactory implements EventFactory<DTask> {

    @Override

    public DTask newInstance() {//disruptor使用环形缓冲区，这是环形缓冲区所承载的对象

        return new DTask();

    }

}

public class DTask {

    public String getName1() {

        return name1;

    }

    public void setName1(String name1) {

        this.name1 = name1;

    }

    public String getName2() {

        return name2;

    }

    public void setName2(String name2) {

        this.name2 = name2;

    }

    public String getName3() {

        return name3;

    }

    public void setName3(String name3) {

        this.name3 = name3;

    }

    String name1;

    String name2;

    String name3;

}

public class DTaskHandle implements EventHandler<DTask> {

    @Override

    public void onEvent(DTask dTask, long l, boolean b) throws Exception {

        System.out.println("开始最后消费");

        System.out.println(dTask.getName1());

        System.out.println(dTask.getName2());

        System.out.println(dTask.getName3());

        System.out.println("结束最后消费");

    }

}

public class DTaskHandle1 implements EventHandler<DTask> {

    @Override

    public void onEvent(DTask dTask, long l, boolean b) throws Exception {

        System.out.println("-----DTaskHandle1-----");

        dTask.setName1("name1");

    }

}

public class DTaskHandle2 implements EventHandler<DTask> {

    @Override

    public void onEvent(DTask dTask, long l, boolean b) throws Exception {

        System.out.println("-----DTaskHandle2-----");

        dTask.setName2("name2");

    }

}

public class DTaskHandle3 implements EventHandler<DTask> {

    @Override

    public void onEvent(DTask dTask, long l, boolean b) throws Exception {

        System.out.println("-----DTaskHandle3-----");

        dTask.setName3("name3");

    }

}

public class DisruptorTest {

    public void exec() throws Exception {

        ExecutorService executor = Executors.newCachedThreadPool();

        Disruptor<DTask> disruptor = new Disruptor(new DTaskFactory(),

                1024 * 1024,

                executor,

                ProducerType.SINGLE, new BusySpinWaitStrategy());

        DTaskHandle dTaskHandle = new DTaskHandle();

        DTaskHandle1 dTaskHandle1 = new DTaskHandle1();

        DTaskHandle2 dTaskHandle2 = new DTaskHandle2();

        DTaskHandle3 dTaskHandle3 = new DTaskHandle3();

        disruptor.handleEventsWith(dTaskHandle1, dTaskHandle2, dTaskHandle3);//消费生产出的对象,并行执行

        disruptor.after(dTaskHandle1, dTaskHandle2, dTaskHandle3).handleEventsWith(dTaskHandle);//并行执行1 2 3后，串行执行dTaskHandle

//        disruptor.

        disruptor.start();

        CountDownLatch latch = new CountDownLatch(1);

        //生产者准备

        executor.submit(new TradePublisher(latch, disruptor));

        latch.await();//等待生产者完事.

        disruptor.shutdown();

        executor.shutdown();

    }

}