本文由 简悦 SimpRead 转码， 原文地址 www.toutiao.com
RingBuffer 如何保证数据不丢失由于 ringbuffer 是一个环形的队列，那么生产者和消费者在遍历这个队列的时候，如何制衡呢？ 1、生产
1、生产快，消费慢，数据丢失？
生产者速度过快，导致一个对象还没消费完，就循环生产了一个新的对象要加入 ringbuffer，导致消费不完整，造成数据丢失？
我们注意到，在我们获取生产者下一个位置的时候，是通过 ringbuffer 的 next 方法，而这个 next 方式是调用了 sequencer 的 next 方法
这个对象，在我们创建 disruptor 对象的时候，创建的
所以这个 ringbuffer 就是 disruptor 中的 sequencer 对象，那么在进行获取 next 的时候，这里是如何获取下一个的呢？是否会对这个生产获取下一个序列进行相应的等待策略，避免产生相应的干扰！！！
这个各位看官还需多看看里面的代码以及封装（特别是封装，真是九转十八弯），多熟悉，我这绕着绕着很容易就绕晕了，刚开始也是云里雾里。
EventProcessor 顾名思义，就是事件处理器（handle 和 process 都可以翻译为 “处理”，但是 process 侧重于机器的处理，而 handle 侧重于有人工的处理，所以使用 handle 表示用户逻辑的处理，使用 process 表示机器的处理），这个接口有两个实现类，分别是 WorkProcessor 和 BatchEventProcessor，它们对应的逻辑处理消费者分别是 EventHandler 和 WorkHandler。下面是 EventProcessor 的 UML 类图及 EventHandler 和 EventProcessor 的接口定义。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
/**
* Callback interface to be implemented for processing events as they become available in the {@link RingBuffer}
*
* @param <T> event implementation storing the data for sharing during exchange or parallel coordination of an event.
* @see BatchEventProcessor#setExceptionHandler(ExceptionHandler) if you want to handle exceptions propagated out of the handler.
* 处理事件的回调接口
*/
public interface EventHandler<T>
{
/**
* Called when a publisher has published an event to the {@link RingBuffer}
*
* @param event published to the {@link RingBuffer}
* @param sequence of the event being processed
* @param endOfBatch flag to indicate if this is the last event in a batch from the {@link RingBuffer}
* @throws Exception if the EventHandler would like the exception handled further up the chain.
*/
void onEvent(T event, long sequence, boolean endOfBatch) throws Exception;
}
/**
* EventProcessors waitFor events to become available for consumption from the {@link RingBuffer}
* <p>
* An EventProcessor will generally be associated with a Thread for execution.
* 事件执行器，等待RingBuffer有可用消费事件。一个事件处理器关联一个执行线程
*/
public interface EventProcessor extends Runnable
{
/**
* Get a reference to the {@link Sequence} being used by this {@link EventProcessor}.
*
* @return reference to the {@link Sequence} for this {@link EventProcessor}
*/
Sequence getSequence();
/**
* Signal that this EventProcessor should stop when it has finished consuming at the next clean break.
* It will call {@link SequenceBarrier#alert()} to notify the thread to check status.
*/
void halt();
boolean isRunning();
}
|
EventProcessor 接口继承了 Runnable 接口，主要有两种实现：单线程批量处理 BatchEventProcessor 和多线程处理 WorkProcessor。
在使用 Disruptor 帮助类构建消费者时，使用 handleEventsWith 方法传入多个 EventHandler，内部使用多个 BatchEventProcessor 关联多个线程执行。这种情况类似 JMS 中的发布订阅模式，同一事件会被多个消费者并行消费。适用于同一事件触发多种操作。
而使用 Disruptor 的
handleEventsWithWorkerPool 传入多个 WorkHandler 时，内部使用多个 WorkProcessor 关联多个线程执行。这种情况类似 JMS 的点对点模式，同一事件会被一组消费者其中之一消费。适用于提升消费者并行处理能力。
我们先回顾下 Disruptor 消费者的两个特点：消费者依赖图（即下文所谓的 “消费链”）和事件多播。
假设现在有 A,B,C,D 四个消费者，它们都能组成什么样的形式呢？从众多的排列组合中，我挑了 4 组比较有代表性的消费链形式。
image.png
- 第 1 组中，消费者 A 消费按成后，B、C、D 可同时消费；
- 第 2 组中，消费者 A、B、C、D 顺序消费；
- 第 3 组中，消费者 A、B 顺序消费后，C、D 同时消费；
- 第 4 组中，消费者 A 在消费完成后，B 和 C 可以同时消费，但是必须在都消费完成后，D 才能消费。
标号为 1、3、4 的消费链都使用了事件多播，可见事件多播属于消费链的一种组合形式。注意，在上面 4 种组合中，每个组合的每一水平行，都属于一个消费者组。
这些还只是较为简单的消费链组成，实际中消费链可能会更复杂。
那么在 Disruptor 内部是怎么实现消费链的呢？
我们可以先思考下。如果想把独立的消费者组成消费链，那么后方的消费者（组）必然要知道在它前方的消费者（组）的处理情况，否则就做不到顺序消费。同时，消费者也要了解生产者的位置，来判断是否有可用事件。之前我们分析生产者代码的时候，已经讲过，生产者为了不覆盖没有消费完全的事件，必须知道最慢消费者的处理情况。
做到了这些才会有能力去控制消费者组成消费链。下面让我们具体看 Disruptor 中的实现。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
package liuqiang.complex.multi;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;
import liuqiang.complex.common.*;
import java.util.concurrent.Executors;
public class Main3 {
//单生产者，多消费者模式。多消费者对于消息不重复消费。例如：1线程消费了消息0，则2线程只能从0后面的消息消费，不能对消息0进行消费。
public static void main(String[] args) throws Exception {
EventFactory<Order> factory = new OrderFactory();
int ringBufferSize = 1024 * 1024;
Disruptor<Order> disruptor =
new Disruptor<Order>(factory, ringBufferSize, Executors.defaultThreadFactory(), ProducerType.SINGLE, new YieldingWaitStrategy());
/*
* 该方法传入的消费者需要实现WorkHandler接口，方法的内部实现是：先创建WorkPool，然后封装WorkPool为EventHandlerPool返回。
* 消费者1、2对于消息的消费有时有竞争，保证同一消息只能有一个消费者消费
*/
disruptor.handleEventsWithWorkerPool(new OrderHandler1("1"), new OrderHandler1("2"));
disruptor.start();
RingBuffer<Order> ringBuffer = disruptor.getRingBuffer();
Producer producer = new Producer(ringBuffer);
//单生产者，生产3条数据
for (int l = 0; l < 3; l++) {
producer.onData(l + "");
}
//为了保证消费者线程已经启动，留足足够的时间。具体原因详见另一篇博客：disruptor的shutdown失效问题
Thread.sleep(1000);
disruptor.shutdown();
}
}
|
调用handleEventsWithWorkerPool 形成 WorkerPool，并进一步封装成 EventHandlerGroup。对于同一条消息，两消费者不重复消费。
可能输出结果如下：
OrderHandler1 1，消费信息：0
OrderHandler1 2，消费信息：1
OrderHandler1 1，消费信息：2
我们重点看一下 SequenceBarrier，可直译为 “序列屏障”。SequenceBarrier 的主要作用是协调获取消费者可处理到的最大序号，内部持有着生产者和其依赖的消费者序列。它的接口定义如下。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
public interface SequenceBarrier
{
/**
* Wait for the given sequence to be available for consumption.<br>
* 等待指定序列可用
* @param sequence to wait for
* @return the sequence up to which is available
* @throws AlertException if a status change has occurred for the Disruptor
* @throws InterruptedException if the thread needs awaking on a condition variable.
* @throws TimeoutException
*
*/
long waitFor(long sequence) throws AlertException, InterruptedException, TimeoutException;
/**
* Get the current cursor value that can be read.<br>
* 获取当前可读游标值
*
* @return value of the cursor for entries that have been published.
*
*/
long getCursor();
/**
* The current alert status for the barrier.<br>
* 当前的alert状态
*
* @return true if in alert otherwise false.
*/
boolean isAlerted();
/**
* Alert the {@link EventProcessor}s of a status change and stay in this status until cleared.<br>
*
* 通知消费者状态变化。当调用EventProcessor#halt()将调用此方法。
*/
void alert();
/**
* Clear the current alert status.<br>
* 清楚alert状态
*/
void clearAlert();
/**
* Check if an alert has been raised and throw an {@link AlertException} if it has.
* 检查是否发生alert，发生将抛出异常
* @throws AlertException if alert has been raised.
*/
void checkAlert() throws AlertException;
}
SequenceBarrier实例引用被EventProcessor持有，用于等待并获取可用的消费事件，主要体现在waitFor这个方法。
|
要实现这个功能，需要 3 点条件：
- 知道生产者的位置。
- 因为 Disruptor 支持消费者链，在不同的消费者组之间，要保证后边的消 费者组只有在前消费者组中的消费者都处理完毕后，才能进行处理。
- 暂时没有事件可消费，在等待可用消费时，还需要使用某种等待策略进行等待。
看下 SequenceBarrier 实现类 ProcessingSequenceBarrier 的代码是如何实现 waitFor 方法。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
final class ProcessingSequenceBarrier implements SequenceBarrier
{
private final WaitStrategy waitStrategy; // 等待可用消费时，指定的等待策略
private final Sequence dependentSequence; // 依赖的上组消费者的序号，如果当前为第一组则为cursorSequence（即生产者发布游标序列），否则使用FixedSequenceGroup封装上组消费者序列
private volatile boolean alerted = false; // 当触发halt时，将标记alerted为true
private final Sequence cursorSequence; // AbstractSequencer中的cursor引用，记录当前发布者发布的最新位置
private final Sequencer sequencer; // MultiProducerSequencer 或 SingleProducerSequencer
public ProcessingSequenceBarrier(
final Sequencer sequencer,
final WaitStrategy waitStrategy,
final Sequence cursorSequence,
final Sequence[] dependentSequences)
{
this.sequencer = sequencer;
this.waitStrategy = waitStrategy;
this.cursorSequence = cursorSequence;
if (0 == dependentSequences.length) // 依赖的上一组序列长度，第一次是0
{
dependentSequence = cursorSequence;
}
else // 将上一组序列数组复制成新数组保存，引用不变
{
dependentSequence = new FixedSequenceGroup(dependentSequences);
}
}
@Override
public long waitFor(final long sequence)
throws AlertException, InterruptedException, TimeoutException
{
// 检查是否停止服务
checkAlert();
// 获取最大可用序号 sequence为给定序号，一般为当前序号+1，cursorSequence记录生产者最新位置，
long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);
if (availableSequence < sequence)
{
return availableSequence;
}
// 返回已发布最高的序列值，将对每个序号进行校验
return sequencer.getHighestPublishedSequence(sequence, availableSequence);
}
// ...
}
|