package com.ozs.da.service.job.impl;

import com.ozs.da.constant.Constant;
import com.ozs.da.handler.ScheduleRoundEventHandler;
import com.ozs.da.handler.ScheduleRoundExceptionHandler;
import com.ozs.da.job.JobServer;
import com.ozs.da.entity.model.ScheduleRoundEvent;
import com.ozs.da.service.job.ScheDisruptInitService;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.Sequence;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.dsl.ProducerType;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.Executors;

/**
 * @author 许明
 * on 2019/3/7.
 * @version 2.0.0
 */
@Service
@Transactional(rollbackFor = Exception.class)
public class ScheDisruptInitServiceImpl implements ScheDisruptInitService {

    private static final Logger logger = LogManager.getLogger(ScheDisruptInitServiceImpl.class);

    @Override
    public void initJobConJudgeDisruptor() {
        JobServer.SCHEDULE_CONDITION_JUDGE_BUFFER = RingBuffer.create(ProducerType.SINGLE, ScheduleRoundEvent.EVENT_FACTORY, 64 * 1024,
                new BlockingWaitStrategy());
        ScheduleRoundEventHandler[] scheRoundEventHandler = new ScheduleRoundEventHandler[Constant.SCHEDU_JOBCONDITION_JUDGE_NUM];
        for (int i = 0; i < Constant.SCHEDU_JOBCONDITION_JUDGE_NUM; i++)
        {
            scheRoundEventHandler[i] = new ScheduleRoundEventHandler();
        }
        WorkerPool<ScheduleRoundEvent> workerPool = new WorkerPool<ScheduleRoundEvent>(JobServer.SCHEDULE_CONDITION_JUDGE_BUFFER,
                JobServer.SCHEDULE_CONDITION_JUDGE_BUFFER.newBarrier(), new ScheduleRoundExceptionHandler(), scheRoundEventHandler);
        Sequence[] sequences = workerPool.getWorkerSequences();
        JobServer.SCHEDULE_CONDITION_JUDGE_BUFFER.addGatingSequences(sequences);
        workerPool.start(Executors.newFixedThreadPool(Constant.SCHEDU_JOBCONDITION_JUDGE_NUM));
        logger.info("调度条件判断线程 处理个数[" + Constant.SCHEDU_JOBCONDITION_JUDGE_NUM + "]");
    }
}
