并发编程框架Disruptor实战 - WaitStrategy消费者等待策略

生产者和消费者都可能出现速度过快,追上对方的情况,这个时候就需要等待了.等待过程中也会有不同的策略.

1 主要策略

当消费者等待在SequenceBarrier上时，有许多可选的等待策略，不同的等待策略在延迟和CPU资源的占用上有所不同，可以视应用场景选择：

1.1 BlockingWaitStrategy

生产者的默认策略是BlockingWaitStrategy,是在RingBuffer中确定的.但是生产者的默认实现MultiProducerSequencer没有使用等待策略

• 生产者是通过LockSupport.parkNanos(1);来等待的
  

  MultiProducerSequencer#next(int n)

  
  消费者的默认策略是
  BatchEventProcessor的run方法
  

  BatchEventProcessor#processEvents()

由SequenceBarrier决定

Disruptor类中

Disruptor#createEventProcessors

SequenceBarrier是由ringBuffer决定的

看一看到最后是由sequencer来决定的

这里的waitStrategy就是sequencer的waitStrategy
默认MultiProducerSequencer

MultiProducerSequencer的默认策略是BlockingWaitStrategy.

sequencer其实就是生产者,所以其实消费者的默认策略和生产者是一样的.

这个策略的内部适用一个锁和条件变量来控制线程的执行和等待（Java基本的同步方法）

最慢的等待策略，但也是CPU使用率最低和最稳定的选项

 public final class BlockingWaitStrategy implements WaitStrategy{    private final Lock lock = new ReentrantLock();    private final Condition processorNotifyCondition = lock.newCondition();    @Override
    public long waitFor(long sequence, Sequence cursorSequence, Sequence dependentSequence, SequenceBarrier barrier)
        throws AlertException, InterruptedException    {        long availableSequence;        // 如果ringBuffer的cursor小于需要的序号，即生产者没有新的事件发出，则阻塞消费者线程，直到生产者通过Sequencer的publish方法唤醒消费者
        if (cursorSequence.get() < sequence)
        {
            lock.lock();            try
            {                while (cursorSequence.get() < sequence)
                {
                    barrier.checkAlert();
                    processorNotifyCondition.await();
                }
            }            finally
            {
                lock.unlock();
            }
        }        // 如果生产者新发布了事件，但是依赖的其他消费者还没处理完，则等待所依赖的消费者先处理
        while ((availableSequence = dependentSequence.get()) < sequence)
        {
            barrier.checkAlert();
            ThreadHints.onSpinWait();
        }        return availableSequence;
    }    @Override
    public void signalAllWhenBlocking()
    {
        lock.lock();        try
        {
            processorNotifyCondition.signalAll();
        }        finally
        {
            lock.unlock();
        }
    }    @Override
    public String toString()
    {        return "BlockingWaitStrategy{" +            "processorNotifyCondition=" + processorNotifyCondition +            '}';
    }
}

1.2 SleepingWaitStrategy

在多次循环尝试不成功后，选择让出CPU，等待下次调度，多次调度后仍不成功，尝试前睡眠一个纳秒级别的时间再尝试
这种策略平衡了延迟和CPU资源占用，但延迟不均匀

1.3 无锁高性能 YieldingWaitStrategy

在多次循环尝试不成功后，选择让出CPU，等待下次调。平衡了延迟和CPU资源占用，但延迟也比较均匀。

作者：JavaEdge
链接：https://www.jianshu.com/p/ba4499de1e83
來源：简书
简书著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。