package testForFun.demo20181210.demo03;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import javax.swing.plaf.synth.SynthOptionPaneUI;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created with IntelliJ IDEA
 * User:Ryannn
 * Date:2018/12/13
 * Time:19:09
 * Test： Disruptor
 */
public class Test01 {

    static class TaskEvent{

        private String taskType;
        private  Integer count;

        public String getTaskType() {
            return taskType;
        }

        public void setTaskType(String taskType) {
            this.taskType = taskType;
        }

        public Integer getCount() {
            return count;
        }

        public void setCount(Integer count) {
            this.count = count;
        }
    }


    public static void main(String[] args) throws InterruptedException {

        ThreadFactory threadFactory = new ThreadFactory() {
            /**
             * Constructs a new {@code Thread}.  Implementations may also initialize
             * priority, name, daemon status, {@code ThreadGroup}, etc.
             *
             * @param r a runnable to be executed by new thread instance
             * @return constructed thread, or {@code null} if the request to
             * create a thread is rejected
             */
            @Override
            public Thread newThread(Runnable r) {
                AtomicInteger integer = new AtomicInteger();
                int num = integer.incrementAndGet();
                System.out.println("生成线程："+num);
                return new Thread(r,"simple Thread"+String.valueOf(num));
            }
        };


        //2,事件工场，初始化事件的时候使用
        EventFactory<TaskEvent> factory = new EventFactory<TaskEvent>() {
            @Override
            public TaskEvent newInstance() {
                return new TaskEvent();
            }
        };

        //3.1，处理PGN事件的handler,
        EventHandler<TaskEvent> handler1 = new EventHandler<TaskEvent>() {
            @Override
            public void onEvent(TaskEvent taskEvent, long l, boolean b) throws Exception {
                if(taskEvent.getTaskType().equals("PGN")){
                    System.out.println("【hander1】处理【PGN事件】"+taskEvent.getCount());
                }
            }
        };


        //3.2,处理PSR事件的hangler
        EventHandler<TaskEvent> handler2 = new EventHandler<TaskEvent>() {
            @Override
            public void onEvent(TaskEvent taskEvent, long l, boolean b) throws Exception {
                if(taskEvent.getTaskType().equals("PSR")){
                    System.out.println("【handler2】处理【PSR事件】"+taskEvent.getCount());
                }
            }
        };
        //4，设置策略
        BlockingWaitStrategy strategy = new BlockingWaitStrategy();

        //5,指定ringBuffer的大小
        int size=1024;

        //6,组装Disruptor
        Disruptor<TaskEvent> disruptor = new Disruptor<>(factory, size, threadFactory, ProducerType.SINGLE, strategy);

        //7,设置处理的hangdler
        disruptor.handleEventsWith(handler1);
        disruptor.handleEventsWith(handler2);

        //8,启动
        disruptor.start();

        //9,测试demo
        RingBuffer<TaskEvent> ringBuffer = disruptor.getRingBuffer();
       //for (int i = 0; i < 3; i++) {
            long sequence = ringBuffer.next();
            try {
                TaskEvent taskEvent = ringBuffer.get(sequence);
              //  taskEvent.setTaskType("PGN");
                taskEvent.setTaskType("PSR");
               // taskEvent.setTaskType("LSR");
                taskEvent.setCount(2);
            }finally {
                ringBuffer.publish(sequence);
            }
            Thread.sleep(10);


    }



}