package com.ozs.da.job;

import com.ozs.da.entity.model.ScheduleRoundEvent;
import com.ozs.da.entity.task.Task;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author 许明
 * on 2019/3/6.
 * @version 2.0.0
 */
public class TestDemo {

    private static Disruptor<ScheduleRoundEvent> disruptor;

    public static void main(String[] args) throws InterruptedException{
        final int BUFFER_SIZE = 1024;
        final int THREAD_NUMBERS = 4;
        EventFactory<ScheduleRoundEvent> eventFactory = new EventFactory<ScheduleRoundEvent>() {
            @Override
            public ScheduleRoundEvent newInstance() {
                return new ScheduleRoundEvent();
            }
        };

//        ExecutorService executors = Executors.newFixedThreadPool(4);
//        disruptor = new Disruptor<ScheduleRoundEvent>(eventFactory, 8 << 8, executors, ProducerType.MULTI, new BlockingWaitStrategy());
//        TestHandler testHandler = new TestHandler();
//        disruptor.handleEventsWith(testHandler);
//        disruptor.start();

        RingBuffer<ScheduleRoundEvent> ringBuffer = RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE);
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        ExecutorService executor = Executors.newFixedThreadPool(THREAD_NUMBERS);
        TestHandler[] workHandlers = new TestHandler[3];
        for (int i = 0; i < 3; i++) {
            TestHandler workHandler = new TestHandler();
            workHandlers[i] = workHandler;
        }
        WorkerPool<ScheduleRoundEvent> workerPool = new WorkerPool<ScheduleRoundEvent>(ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(), workHandlers);
        workerPool.start(executor);


//        RingBuffer<ScheduleRoundEvent> ringBuffer1 = disruptor.getRingBuffer();
        for (int i = 0; i < 100; i++) {
            long seq = ringBuffer.next();
            Task task = new Task();
            task.setTaskName("测试任务名称"+ i);
            ringBuffer.get(seq).setTask(task);
            ringBuffer.publish(seq);
//            if (i % 10 == 0) {
//                System.out.println(((ThreadPoolExecutor) executor).getActiveCount() + "============================================");
//            }
        }
//        Thread.sleep(1000);
//        workerPool.halt();
//        Thread.sleep(1);
//        executor.shutdown();

    }

}
