Disruptor结构图.png
disruptor的核心设计思路
- 使用循环数组的方式代替队列，使用预先填充数据的方式来避免GC
- 使用CPU缓存行填充的方式来避免极端情况下的数据争用导致的性能下降
- 多线程编程中尽量避免锁争用的编码技巧
生成端减少资源的竞争
多线程在队列也好，循环数组也好，必然存在对标志位的竞争。无论是使用锁来避免竞争，还是使用cas来进行无锁算法。只要争用的情况存在，并且线程较多，都会出现对资源的不断消耗。争用的对象越多，争用中消耗掉的资源也就越多。为了避免这样的情况，减少争用的资源就是一个手段。比如在循环数组中只保留一个标志位，也就是下一个可以写入数据位置的标志位。而尾部标志位则在各个消费者线程中保存（具体的编程手法后续细讲）
如果存在多个生产者，则可写入的标志位需要用cas算法来进行争夺，避免锁的使用。多个线程通过cas得到唯一的不冲突的下一个可写序号。由于需要获得序号后才能进行写入，而写入完成才可以让消费者线程进行消费。所以才获得序号后，完成写入前，必须有一种方式让消费者检测是否完成。以避免消费者拿到还未填入输入的数组位。
具体编程手法
主要的争夺环节集中在多线程发布中，序号大的线程发布需要等到序号小的线程发布完成后才能发布。那我们的优化的点也在这个地方。如果只有一个地方可以写入完成信息，必然需要争夺。为了避免争夺，我们可以使用标志数组（长度和内容数组相同，每一位表示相同下标的内容数组是否发布）来表示每一个位置是否写入。这样就可以避免发布的争夺（大家的标志位都不在一起了）。这也是Disruptor里面的availableSequence
但是又来带来一个问题，用什么数字来表示是否已经发布完成？如果只是0和1，那么写过1轮以后，标志数组位上就都是1了。又无法区分。所以标志数组上的数字应该在循环数组的每一轮循环的值都不同。比如一开始都是-1，第一轮中是0的表示已发布，第二轮中是0表示没发布，是1的表示已发布。
Disruptor中的具体处理
发布的算法步骤
1.将序号除以标志数组长度（因为长度是2的次方幂，这一步可以通过右移来完成）得到填入值x。(序号每次都会增长,所以每次都应该不同)
2.将序号和标志数组长度减一进行并运算得到填入位置index。即类似于hash&(length-1)获取填入位置
4.将x写入index位置。
消费端减少对锁的使用
生产者消费者中最容易出现争夺的,采用的优化手段包括
- 使用循环数组代替队列，使用cas算法来代替锁争夺
- 消费者各自保存自己当前已经处理过的序号，而不是将这个序号的信息在队列中来存储，避免多线程争用。
- 生产者线程则需要持有消费者的类的信息，好用来判断所有消费者中消费的最小的序号，以避免在数据写入时覆盖了某个消费者尚未处理的数据信息。(生成端是怎么来获取消费者消费最小的序号的,后面讲)
伪共享
这个主要跟计算机的基本结构。L1、L2、L3分别表示一级缓存、二级缓存、三级缓存，越靠近CPU的缓存，速度越快，容量也越小。所以L1缓存很小但很快，并且紧靠着在使用它的CPU内核；L2大一些，也慢一些，并且仍然只能被一个单独的CPU核使用；L3更大、更慢，并且被单个插槽上的所有CPU核共享；最后是主存，由全部插槽上的所有CPU核共享。
当CPU执行运算的时候，它先去L1查找所需的数据、再去L2、然后是L3，如果最后这些缓存中都没有，所需的数据就要去主内存拿。走得越远，运算耗费的时间就越长。所以如果你在做一些很频繁的事，你要尽量确保数据在L1缓存中。
如果不同的线程变量处于同一个缓存行,及时其中一个的缓存数据是最新的,如果另外一个变量需要更新,那么整个缓存行都会更新,具体可以参见计算机MESI协议.
解决伪共享的一种方法是通过补齐(Padding)，使得每一条缓存行只存一个多线程变量
32位的计算机上一个缓存行是64个字节,而一个Java的long类型变量是8字节，因此在一个缓存行中可以存8个long类型的变量
CPU每次从主存中拉取数据时，会把相邻的数据也存入同一个cache line
在访问一个long数组的时候，如果数组中的一个值被加载到缓存中，它会自动加载另外7个
所以一般都是左右补7个。但这种方法是有局限性就是和计算机位数有关系以及jvm可能会对无用的变量做优化导致补齐并不能生效
解决伪共享的第二种方法就是使用@ Contended注解,这个是java给我们提供的方案
下面看源码
单生产者生产数据
1.申请写入m个元素
2.若是有m个元素可以写入，则返回最大的序列号。这儿需要注意的是是否会覆盖未读的元素
3.若是返回正确，则生产者开始写入元素
public long next(int n)
{
if (n < 1)
{
throw new IllegalArgumentException("n must be > 0");
}
long nextValue = this.nextValue; 上一次生产的
long nextSequence = nextValue + n;
long wrapPoint = nextSequence - bufferSize; 同一起跑线进行比较
long cachedGatingSequence = this.cachedValue;
if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue)
{
cursor.setVolatile(nextValue); // StoreLoad fence
long minSequence;
判断是否覆盖了最慢的消费者
while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences, nextValue)))
{
LockSupport.parkNanos(1L); // TODO: Use waitStrategy to spin? 进行阻塞
}
this.cachedValue = minSequence; 将最慢的消费者缓存起来
}
this.nextValue = nextSequence;
return nextSequence;
}
image.png
多个生产者
防止多个线程重复写同一个元素
解决方案:
- 每个线程获取不同的一段数组空间进行操作,在分配元素的时候，通过CAS判断一下这段空间是否已经分配出去即可
防止读取的时候，读到还未写的元素
解决方案
引入了一个与Ring Buffer大小相同的buffer
即位数组available Buffer
当某个位置写入成功的时候，便把availble Buffer相应的位置置位，标记为写入成功
读取的时候，会遍历available Buffer，来判断元素是否已经就绪
代码如下,引入了CAS操作以及原子变量
public long next(int n)
{
if (n < 1)
{
throw new IllegalArgumentException("n must be > 0");
}
long current;
long next;
do
{
current = cursor.get();
next = current + n;
long wrapPoint = next - bufferSize;
long cachedGatingSequence = gatingSequenceCache.get();
if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current)
{
long gatingSequence = Util.getMinimumSequence(gatingSequences, current); 通过这个方法获取到消费端最慢的序号
if (wrapPoint > gatingSequence)
{
LockSupport.parkNanos(1); // TODO, should we spin based on the wait strategy?
continue;
}
gatingSequenceCache.set(gatingSequence);
}
else if (cursor.compareAndSet(current, next))
{
break;
}
}
while (true);
return next;
}
再来看消费者是怎么消费的
单消费者
com.lmax.disruptor.dsl.Disruptor.start
for循环
->com.lmax.disruptor.dsl.EventProcessorInfo.start
->com.lmax.disruptor.BatchEventProcessor.run
BatchEventProcessor.run()方法如下
public void run()
{
if (running.compareAndSet(IDLE, RUNNING))
{
sequenceBarrier.clearAlert();
notifyStart();
try
{
if (running.get() == RUNNING)
{
processEvents();
}
}
finally
{
notifyShutdown();
running.set(IDLE);
}
}
else
{
// This is a little bit of guess work. The running state could of changed to HALTED by
// this point. However, Java does not have compareAndExchange which is the only way
// to get it exactly correct.
if (running.get() == RUNNING)
{
throw new IllegalStateException("Thread is already running");
}
else
{
earlyExit();
}
}
}
processEvents()方法如下
private void processEvents()
{
T event = null;
long nextSequence = sequence.get() + 1L;
while (true)
{
try
{
final long availableSequence = sequenceBarrier.waitFor(nextSequence); 从位数组中获取可消费的序号
if (batchStartAware != null)
{
batchStartAware.onBatchStart(availableSequence - nextSequence + 1); 获取做大的可消费序号,批量消费
}
while (nextSequence <= availableSequence)
{
event = dataProvider.get(nextSequence);
eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence); 调用消费者消费的逻辑
nextSequence++;
}
sequence.set(availableSequence);
}
catch (final TimeoutException e)
{
notifyTimeout(sequence.get());
}
catch (final AlertException ex)
{
if (running.get() != RUNNING)
{
break;
}
}
catch (final Throwable ex)
{
exceptionHandler.handleEventException(ex, nextSequence, event);
sequence.set(nextSequence);
nextSequence++;
}
}
}
多消费者情况
com.lmax.disruptor.dsl.Disruptor.start
for循环
->com.lmax.disruptor.dsl.EventProcessorInfo.start
->com.lmax.disruptor.WorkProcessor.run
WorkProcessor.run()方法
public void run()
{
if (!running.compareAndSet(false, true))
{
throw new IllegalStateException("Thread is already running");
}
sequenceBarrier.clearAlert();
notifyStart();
boolean processedSequence = true;
long cachedAvailableSequence = Long.MIN_VALUE;
long nextSequence = sequence.get(); 使用了原子变量
T event = null;
while (true)
{
try
{
// if previous sequence was processed - fetch the next sequence and set
// that we have successfully processed the previous sequence
// typically, this will be true
// this prevents the sequence getting too far forward if an exception
// is thrown from the WorkHandler
使用CAS来获取可消费的序号
if (processedSequence)
{
processedSequence = false;
do
{
nextSequence = workSequence.get() + 1L;
sequence.set(nextSequence - 1L);
}
while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));
}
if (cachedAvailableSequence >= nextSequence)
{
event = ringBuffer.get(nextSequence);
workHandler.onEvent(event);
processedSequence = true;
}
else
{
cachedAvailableSequence = sequenceBarrier.waitFor(nextSequence);
}
}
catch (final TimeoutException e)
{
notifyTimeout(sequence.get());
}
catch (final AlertException ex)
{
if (!running.get())
{
break;
}
}
catch (final Throwable ex)
{
// handle, mark as processed, unless the exception handler threw an exception
exceptionHandler.handleEventException(ex, nextSequence, event);
processedSequence = true;
}
}
notifyShutdown();
running.set(false);
}
所以,单消费者和多消费者的区别是什么？
单消费者时:
1.直接获取欲发布的
long nextSequence = sequence.get() + 1L;
2.满足条件时进行数据处理
while (nextSequence <= availableSequence)
{
event = dataProvider.get(nextSequence);
eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);
nextSequence++;
}
3.更新序号
sequence.set(availableSequence);
这里是不需要考虑并发问题的
多消费者时
消费者保持一个自己的序列，每次累加后nextSequence，去获取可访问的最大序列。对于一个生产者，就是nextSequence到RingBuffer当前游标的序列。对于多个生产者，就是nextSequence到RingBuffer当前游标之间，最大的连续的序列集。
1.使用cas操作获取欲发布的
do
{
nextSequence = workSequence.get() + 1L;
sequence.set(nextSequence - 1L);//备注 直接在这里更新序号了
}
while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));
2.满足条件时进行数据处理
if (cachedAvailableSequence >= nextSequence)
{
event = ringBuffer.get(nextSequence);
workHandler.onEvent(event);
processedSequence = true;
}
3.更新序号
消费者的等待策略
| 名称 | 措施 | 适用场景 |
|---|---|---|
| BlockingWaitStrategy | 加锁 | CPU资源紧缺，吞吐量和延迟并不重要的场景 |
| BusySpinWaitStrategy | 自旋 | 通过不断重试，减少切换线程导致的系统调用，而降低延迟。推荐在线程绑定到固定的CPU的场景下使用 |
| PhasedBackoffWaitStrategy | 自旋 + yield + 自定义策略 | CPU资源紧缺，吞吐量和延迟并不重要的场景 |
| SleepingWaitStrategy | 自旋 + yield + sleep | 性能和CPU资源之间有很好的折中。延迟不均匀 |
| TimeoutBlockingWaitStrategy | 加锁，有超时限制 | CPU资源紧缺，吞吐量和延迟并不重要的场景 |
| YieldingWaitStrategy | 自旋 + yield + 自旋 | 性能和CPU资源之间有很好的折中。延迟比较均匀 |
网友评论