import com.lmax.disruptor.LiteBlockingWaitStrategy; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public static void main(String[] args) throws InterruptedException {
	//Executor that will be used to construct new threads for consumers
	Executor executor = Executors.newCachedThreadPool();
	//Specify the size of the ring buffer, must be power of 2.
	int bufferSize = 1024;
	//Disruptor<ObjectEvent> disruptor = new Disruptor<>(ObjectEvent::new, bufferSize, executor);
	Disruptor<ObjectEvent> disruptor = new Disruptor<>(ObjectEvent::new, bufferSize, executor, 
			ProducerType.SINGLE, new LiteBlockingWaitStrategy());

	disruptor.handleEventsWith(App::handleEvent1);
	disruptor.handleEventsWith(App::handleEvent2);
	
	//disruptor.handleEventsWith((event, sequence, endOfBatch) -> System.out.println("Event: " + event.getObject()));
	disruptor.start();

	produceEvents(disruptor);
}
 

import com.lmax.disruptor.LiteBlockingWaitStrategy; //导入依赖的package包/类
/**
 * Creates a new MetricsMetaAPIImpl
 * @param properties The configuration properties
 */
public MetricsMetaAPIImpl(final Properties properties) {
	dataSource = SQLCompilerDataSource.getInstance(properties);
	sqlWorker = dataSource.getSQLWorker();
	tagPredicateCache = new TagPredicateCache(sqlWorker);
	fjPool = new ManagedForkJoinPool(getClass().getSimpleName(), Runtime.getRuntime().availableProcessors(), true, JMXHelper.objectName(getClass()));
	metaReader = new DefaultMetaReader(sqlWorker);
	dispatcher = new WorkQueueDispatcher("MetricsMetaDispatcher", Runtime.getRuntime().availableProcessors(), 1024, this, ProducerType.MULTI, new LiteBlockingWaitStrategy());
	log.info("Dispatcher Alive: {}", dispatcher.alive());
}
 

import com.lmax.disruptor.LiteBlockingWaitStrategy; //导入依赖的package包/类
/**
 * 根据config初始化特殊通道
 *
 * @param symbol    事件
 * @param listeners 对应的监听器集合
 */
private void initSpecDisruptor(String symbol, List<ElectronsListener> listeners) {
    ExecutorService specPool = Executors.newFixedThreadPool(conf.getSpecCircuitNum(), new ThreadFactory() {

        final AtomicInteger cursor = new AtomicInteger(0);

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, "Electrons Thread (from spec channel) : thread" + cursor.incrementAndGet());
        }
    });
    pools.add(specPool);

    Disruptor<ElectronsHolder> disruptor = new Disruptor<>(ElectronsHolder::new, conf.getSpecCircuitLen(), specPool, ProducerType.MULTI, new LiteBlockingWaitStrategy());
    disruptor.handleExceptionsWith(new ElecExceptionHandler("Spec Disruptor {" + symbol + "}"));

    //初始化管道并放入集合中
    SpecChannel specChannel = new SpecChannel(disruptor);
    if (conf.isBreaker()) {
        EventCountCircuitBreaker breaker = new EventCountCircuitBreaker(conf.getErrorNum(), conf.getPerUnit(), conf.getUnit(), conf.getCloseThreshold(), conf.getRest(), conf.getRestUnit());
        specChannel.setBreaker(breaker);
    }

    //构建listener顺序
    ListenerChainBuilderNew.buildChain(specChannel, listeners);

    channelMap.put(SPEC_CHANNEL_PREFIX + symbol, specChannel);
}
 

import com.lmax.disruptor.LiteBlockingWaitStrategy; //导入依赖的package包/类
/**
 * 初始化正常管道，任何情况下都会有
 *
 * @param pool 线程池
 */
private void initNormalChannel(ExecutorService pool) {
    Disruptor<ElectronsHolder> normalDis = new Disruptor<>(ElectronsHolder::new, conf.getCircuitLen(), pool, ProducerType.MULTI, new LiteBlockingWaitStrategy());
    WorkHandler[] workHandlers = new WorkHandler[conf.getCircuitNum()];
    Arrays.fill(workHandlers, (WorkHandler<ElectronsHolder>) electronsHolder -> electronsHolder.handle());
    normalDis.handleEventsWithWorkerPool(workHandlers);
    normalDis.handleExceptionsWith(new ElecExceptionHandler("Normal Disruptor"));

    //初始化channel
    Channel normalChannel = new NormalChannel(normalDis);
    //配置限流相关
    normalChannel.confLimitRate(conf.isLimitRate(), conf.getPermitsPerSecond(), conf.isWarmup(), conf.getWarmupPeriod(), conf.getWarmPeriodUnit());
    channelMap.put(NORMAL_CHANNEL_KEY, normalChannel);
}
 

import com.lmax.disruptor.LiteBlockingWaitStrategy; //导入依赖的package包/类
@Test
public void test_All_WaitStrategies() {
	assertTrue(WaitStrategyType.BLOCKING.instance() instanceof BlockingWaitStrategy);
	assertTrue(WaitStrategyType.BUSY_SPIN.instance() instanceof BusySpinWaitStrategy);
	assertTrue(WaitStrategyType.LITE_BLOCKING.instance() instanceof LiteBlockingWaitStrategy);
	assertTrue(WaitStrategyType.SLEEPING_WAIT.instance() instanceof SleepingWaitStrategy);
	assertTrue(WaitStrategyType.YIELDING.instance() instanceof YieldingWaitStrategy);
}
 

import com.lmax.disruptor.LiteBlockingWaitStrategy; //导入依赖的package包/类
/**
 * Create a new RingBufferProcessor using {@link #SMALL_BUFFER_SIZE} backlog size, blockingWait Strategy
 * and auto-cancel.
 * <p>
 * A new Cached ThreadExecutorPool will be implicitely created.
 *
 * @param <E> Type of processed signals
 * @return a fresh processor
 */
public static <E> RingBufferProcessor<E> create() {
  return create(RingBufferProcessor.class.getSimpleName(), SMALL_BUFFER_SIZE, new LiteBlockingWaitStrategy(), true);
}
 

import com.lmax.disruptor.LiteBlockingWaitStrategy; //导入依赖的package包/类
/**
 * Create a new RingBufferProcessor using {@link #SMALL_BUFFER_SIZE} backlog size, blockingWait Strategy
 * and auto-cancel.
 * <p>
 * A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that
 * will fan-in data.
 * <p>
 * A new Cached ThreadExecutorPool will be implicitely created.
 *
 * @param <E> Type of processed signals
 * @return a fresh processor
 */
public static <E> RingBufferProcessor<E> share() {
  return share(RingBufferProcessor.class.getSimpleName(), SMALL_BUFFER_SIZE, new LiteBlockingWaitStrategy(), true);
}
 

import com.lmax.disruptor.LiteBlockingWaitStrategy; //导入依赖的package包/类
/**
 * Create a new RingBufferWorkProcessor using {@link #SMALL_BUFFER_SIZE} backlog size, blockingWait Strategy
 * and auto-cancel.
 * <p>
 * A new Cached ThreadExecutorPool will be implicitely created.
 *
 * @param <E> Type of processed signals
 * @return a fresh processor
 */
public static <E> RingBufferWorkProcessor<E> create() {
  return create(RingBufferWorkProcessor.class.getSimpleName(), SMALL_BUFFER_SIZE, new
    LiteBlockingWaitStrategy(), true);
}
 

import com.lmax.disruptor.LiteBlockingWaitStrategy; //导入依赖的package包/类
/**
 * Create a new RingBufferWorkProcessor using {@link #SMALL_BUFFER_SIZE} backlog size, blockingWait Strategy
 * and auto-cancel.
 * <p>
 * A Shared Processor authorizes concurrent onNext calls and is suited for multi-threaded publisher that
 * will fan-in data.
 * <p>
 * A new Cached ThreadExecutorPool will be implicitely created.
 *
 * @param <E> Type of processed signals
 * @return a fresh processor
 */
public static <E> RingBufferWorkProcessor<E> share() {
  return share(RingBufferWorkProcessor.class.getSimpleName(), SMALL_BUFFER_SIZE, new
    LiteBlockingWaitStrategy(), true);
}