RingBuffer 源码解析

RingBuffer 主要用于在生产者和消费者之间传递数据。至于核心的控制机制已经被转移到 Sequencer 中了,具体的控制机制后续会有写专门的文章,这里先聚焦 RingBuffer。

RingBuffer 基本结构

缓存行填充技术在 Disruptor 中被大量使用,在这里同样也用到了,关于缓存行填充在上一篇文章中已经姜的很详细了,就是为了将 RingBuffer 中不会修改的值一直缓存在 CPU 缓存中,减少访问内存的次数。

而且 RingBuffer 中的对象在 Disruptor 启动时就会被创建后,后续会对这些对象进行重复利用,而不用创建新的对象,所以 Disruptor 中也基本不用进行垃圾回收,以下的代码就是用来初始化对象:

1
2
3
4
5
private void fill(EventFactory<E> eventFactory){    
for (int i = 0; i < bufferSize; i++)    {       
entries[BUFFER_PAD + i] = eventFactory.newInstance();   
}
}

在初始化 RingBuffer 时,需要指定三个参数,EventFactory,bufferSize 和 WaitStrategy。EventFactory 就是用来生成数组中元素的工厂,如上代码。

bufferSize 是用来存储消息的数组的容量, 必须是 2 的次方,在 Java 的 HashMap 中,也需要保证容量是 2 的次方,这两个地方的设计是相同的,都是为了可以快速找到目标地址。

我们知道 RingBuffer 中是使用循环数组来存储消息,获取元素位置的时候一般会使用 & 来获取元素位置,如下:

1
2
3
protected final E elementAt(long sequence){   
return (E) UNSAFE.getObject(entries, REF_ARRAY_BASE + ((sequence & indexMask) << REF_ELEMENT_SHIFT));
}

sequence 就是当前消息的序列号,indexMask 的值是 bufferSize - 1。在 bufferSize 的值为 2 的 N 次方的情况下,sequence & indexMask 的值就是元素的位置。在 HashMap 中,如果传入的值不是 2 的 N 次方,会被自动转成最接近传入值的 2 的 N 次方,而在 Disruptor 中,则会直接报错。

其实利用取余操作也可以获取到元素的位置,但是取余的操作效率是比 & 操作效率低,对于 Disruptor 这么最求性能极致的框架来说,怎么会放过这个压榨性能的机会。

WaitStrategy 是消费者在等待 RingBuffer 中消息的策略,默认是 BlockingWaitStrategy,也就是获取不到消息时进行阻塞,也可以根据的自己的需要实现自己的等待策略。

创建 RingBuffer 时,可以指定是一个生产者还是多个生产者,但实际就是创建不同的 Sequencer,Sequencer 的原理我们下篇文章再讲:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// 生成支持多个生产者的 RingBuffer, 实际是创建不同的 Sequencer
public static <E> RingBuffer<E> createMultiProducer( EventFactory<E> factory, int bufferSize, WaitStrategy waitStrategy)
{
MultiProducerSequencer sequencer = new MultiProducerSequencer(bufferSize, waitStrategy);

return new RingBuffer<E>(factory, sequencer);
}

// 生成支持单个生产者的 RingBuffer
public static <E> RingBuffer<E> createSingleProducer(EventFactory<E> factory, int bufferSize, WaitStrategy waitStrategy)
{
SingleProducerSequencer sequencer = new SingleProducerSequencer(bufferSize, waitStrategy);

return new RingBuffer<E>(factory, sequencer);
}

除了上面两种默认的方式之外,还可以通过 create 方法来创建指定类型的 RingBuffer。

RingBuffer 的核心功能就是存储和传输消息,其他功能已经被抽离到其他组件中。

RingBuffer 如何传输数据

RingBuffer 的核心功能就是将消息从生产者取过来,然后给消费者处理。

生产者生产数据

在 RingBuffer 中有很多 translateAndPublish 重载方法,支持不同的参数,但是核心都是传入一个 EventTranslator 的实现类,并且需要实现 translateTo 方法,也就是说,如何将数据放进 RingBuffer 操作又回到了用户的手上。

1
2
3
4
5
6
7
8
9
10
11
private <A> void translateAndPublish(EventTranslatorOneArg<E, A> translator, long sequence, A arg0)
{
try
{
translator.translateTo(get(sequence), sequence, arg0);
}
finally
{
sequencer.publish(sequence);
}
}

简单的 EventTranslator 实现如下:

1
2
3
4
5
6
7
8
EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR =
new EventTranslatorOneArg<LongEvent, ByteBuffer>()
{
public void translateTo(LongEvent event, long sequence, ByteBuffer bb)
{
event.set(bb.getLong(0));
}
};

除了利用 EventTranslator,我们也可以自己来实现完整的生产者,如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public class LongEventProducer {
private final RingBuffer<LongEvent> ringBuffer;

public LongEventProducer(RingBuffer<LongEvent> ringBuffer)
{
this.ringBuffer = ringBuffer;
}

public void onData(ByteBuffer bb)
{
long sequence = ringBuffer.next();
try
{
LongEvent event = ringBuffer.get(sequence);
event.set(bb.getLong(0));
}
finally
{
ringBuffer.publish(sequence);
}
}
}

对比这两种不同的实现可以发现,在将消息存入 RingBuffer 时,需要获取一个 sequence,并且在存入数据后需要调用 publish 方法。

获取 sequence 表示的就是当前可以存入数据的位置,在存入数据之后,需要通过 publish 方法高速消费者这个位置的消息可用。

消费者处理数据

在实例化 Disruptor 时,需要配置 Handler:

1
2
Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(factory, bufferSize, DaemonThreadFactory.INSTANCE);
disruptor.handleEventsWith(new LongEventHandler());

同样,数据具体如何处理,也由用户自己来决定:

1
2
3
4
5
6
public class LongEventHandler implements EventHandler<LongEvent> {
public void onEvent(LongEvent event, long sequence, boolean endOfBatch)
{
System.out.println("Event: " + event);
}
}

Disruptor 支持多个消费者,所以可以添加多个消费者:

1
disruptor.handleEventsWith(new LongEventHandler(), new AnotherLongEventHandler());

当然也可以调整这几个 Handler 之间的执行顺序:

1
disruptor.handleEventsWith(new LongEventHandler()).then(new AnotherLongEventHandler());

随着 Disruptor 的演进,RingBuffer 功能也慢慢变的单一,只要负责数据的存储和传输,不需要负责具体的控制逻辑,这样整个框架结构更je简单,内聚性更好,这个符合 Disruptor 追求极致性能的目标。

© 2020 Rayjun    PowerBy Hexo