RingBuffer 在 Puma 中的应用

什么是 RingBuffer

环形缓冲区：https://zh.wikipedia.org/wiki/環形緩衝區

维基百科的解释是：它是一种用于表示一个固定尺寸、头尾相连的缓冲区的数据结构，适合缓存数据流。

底层数据结构很是简单，一个固定长度的数组加一个写指针和一个读指针。

只要像这张图同样，把这个数组辦弯，它就成了一个 RingBuffer。

那它到底有什么精妙的地方呢？

我最近作的项目正好要用到相似的设计思路，因此翻出了之前在点评写的 Puma 系统，看了看之前本身写的代码。顺便写个文章总结一下。

Puma 是什么，为何要用 RingBuffer

Puma 简介

Puma 是一个 MySQL 数据库 Binlog 订阅消费系统。相似于阿里的 Canel。

Puma 会假装成一个 MySQL Slave，而后消费 Binlog 数据，并缓存在本地。当有客户端链接上来的时候，就会从本地读取数据给客户端消费。

若是用很通常的设计，一个单独的线程会从 MySQL 消费数据，存到本地文件中。而后每一个客户端的链接都会有一个线程，从本地文件中读取数据。

没错，初版就是这么简单粗暴，固然，它是有效的。

利用缓存优化性能

不用作性能测试就知道，系统压力大了之后这里一定会是一个瓶颈。读写数据会有一点延时，若是多个客户端同时读取同一份数据又会形成不少的浪费。而后数据的编码解码还会有很多损耗。

因此这里固然要加一个缓存了。

一个线程写数据到磁盘，而后它能够同时把数据传递给各个客户端。

听起来像发布者订阅者模式？也有点像生产者消费者模式？

可是，它并不只仅是发布者订阅者模式，由于这里的“发布者”和“订阅者”是彻底异步的，并且每一个“订阅者”的消费速度是不同的。

它也不只仅是生产者消费者模式，由于这里的“消费者”是同时消费全部数据，而不是把数据分发给各个“消费者”。

它们的消费速度不同，还会出现缓存内的数据过新，“消费者”不得不去磁盘读取。

感受这里的需求是两种设计模式的结合。

不只如此，这是一套高并发的系统，怎么保证性能，怎么保证数据一致性？

因此，这个缓存看上去简单，其实它不简单。

常规解决思路

目标明确后，看看 Java 的并发集合中有什么能知足需求的吧。

第一个想到的就是BlockingQueue，为每个客户端建立一个BlockingQueue，BlockingQueue内部是经过加锁来实现的，虽然锁冲突不会不少，但高并发的状况下，最好仍是能作到无锁。

Disruptor

当时正好看到了一系列介绍 Disruptor 的文章：传送门

因此就在想能不能把 RingBuffer 用来解决咱们的问题呢？

对 RingBuffer 进行改进

看完了 RingBuffer 的基本原理后，就要开始用它来适应咱们的系统了。这里遇到了几个问题：

1. 如何支持多个消费者

2. 如何判断当前有无新数据，如何判断当前数据是否已经被新数据覆盖

3. 如何保证数据一致性

第一个问题

这个问题简单，原始的 RingBuffer 只有一个写指针和一个读指针。

要支持多个消费者的话，只要为每一个消费者建立一个读指针便可。

第二个问题

RingBuffer 的一个精髓就是，写指针和读指针的大小是会超过数组长度的，写入和读取数据的时候，是采用writeIndex % CACHED_SIZE这样的形式来读取的。

为何要这么作？这就是为了解决判断有无新数据和数据是否已被覆盖的问题。

假设我内部2个指针，分别叫nextWriteIndex，nextReadIndex。

那么判断有无新数据的逻辑就是if (nextReadIndex >= nextWriteIndex)，返回true的话就是没有新数据了。

而判断数据是否被覆盖的逻辑就是if (nextReadIndex < nextWriteIndex - CACHED_SIZE)，返回true的话就是数据已经被覆盖了。

拿实际数据举个例子：

一个长度为10的 RingBuffer，内部是一个长度为10的数组。

此时nextWriteIndex=12，意味着它下一次写入的数据会在 12%10=2 上。此时，可读的有效范围是 2~11，对应的数组内的索引就是 2, 3, 4, 5, 6, 7, 8, 9, 0, 1。

因此，当nextReadIndex=12 的时候，会读到最老的数据2，这是老数据，不是新数据，此时表示没有行数据了。

当nextReadIndex＝1 的时候，是新数据，而不是想要的老数据，老数据已经被覆盖掉了，此时它没办法从缓存里读数据了。

第三个问题

最棘手的第三个问题来了，这个系统是要支持高并发的，若是是同步的操做，上面的代码没有任何问题。或者说，若是是同步的代码，干吗还要用 RingBuffer 呢？

上面写入和读取，都有两步操做，更改数据和更改索引，按照逻辑上来说，它们应该是强一致性的。只能加锁了？若是要加锁，为什么不直接用BlockingQueue？

因此，是否能够经过什么方法，高并发和最终一致性呢？

直接贴代码吧，根据代码一步步分析：

public class CachedDataStorage {

    private static final int CACHED_SIZE = 5000;

    private final ChangedEventWithSequence[] data = new ChangedEventWithSequence[CACHED_SIZE];

    private volatile long nextWriteIndex = 0;

    public void append(Object dataValue) {
        data[(int) (nextWriteIndex % CACHED_SIZE)] = dataValue;
        nextWriteIndex++;
    }

    public Reader createReader() {
        return new Reader();
    }

    public class Reader {

        private Reader() {
        }

        private volatile long nextReadIndex = 0;

        public Object next() throws IOException {
            if (nextReadIndex >= nextWriteIndex) {
                return null;
            }

            if (nextReadIndex <= nextWriteIndex - CACHED_SIZE) {
                throw new IOException("data outdated");
            }

            Object dataValue = data[(int) (nextReadIndex % CACHED_SIZE)];

            if (nextReadIndex <= nextWriteIndex - CACHED_SIZE) {
                throw new IOException("data outdated");
            } else {
                nextReadIndex++;
                return dataValue;
            }
        }
    }
}

咱们来一步步分析，先看内部的Reader和createReader()方法，每来一个客户端就会建立一个Reader，每一个Reader会维护一个nextReadIndex。

而后看append()方法，能够说没有任何逻辑，直接写入数据，修改索引就结束了。可是，别小看了这两个步骤的操做顺序。

好了，到了最复杂的next()方法了，这里可就大有讲究了。

一进来马上执行if (nextReadIndex >= nextWriteIndex)，用来判断当前是否还有更新的数据。

由于写入的时候是先写数据再改索引，因此可能会出现明明有数据，可是这里认为没数据的状况。

可是并无关系，咱们更关注最终一致性，由于咱们要的是确保这里必定不会读错数据，而不必定要确保这里有新数据就要马上处理。就算这一轮没读到，下一轮也必定会读取到了。

下一步是这一行if (nextReadIndex <= nextWriteIndex - CACHED_SIZE)，判断想要读取的数据有没有被新数据覆盖。等一下，这里为何和上面介绍的不同？

上面写的是<，而这里倒是<=。上面提到，同步操做的状况下，用<是没有问题的，可是这里的异步的。

写入数据的时候，可能会出现数据已被覆盖，而索引未被更新的问题，因此这样子判断能够保证不会读错数据。

既然上下边界都检查过了，那么就读取数据吧！就当这里准备读数据的时候，写数据的线程居然又写入了好多数据，致使读出来的数据已经被覆盖了！

因此，必定要在读完数据后，再次检查数据是否被覆盖。

最终，整个过程实现了无锁，高并发和最终一致性。

在 Puma 系统中，启用缓存和关闭缓存，一写五读的状况下，性能整整提升了一倍。测试仍是在我 SSD 上进行的，若是是传统硬盘，提高会更明显。

利用 RingBuffer 实现后的优势

代码实现完，就能够和BlockingQueue对比一下了。

首先，RingBuffer 彻底是无锁的，没有任何锁冲突。而利用BlockingQueue的话它内部会加锁，虽然锁冲突不会不少，可是没锁确定比有锁好。并且，当 Writer 往多个BlockingQueue中顺序写入数据的时候，会有相互影响。而利用 RingBuffer 实现的话，不管有多少 Reader，都不会影响写入性能。

而后是内存上的优点，每次多一个 Reader 仅仅是多一个对象，Reader对象内部也只有一个变量，占用内存很是很是小。而使用BlockingQueue的话，须要建立的就不只仅是一个对象了，会有一系列的东西。

目前看来，该实现能知足咱们的需求且无明显缺点，并且已经在系统中平稳运行了将近一年了。在我离职以前的最后一个项目，就包括为点评订单系统接入 Puma，订单系统在活动期间的写入 QPS 很是高。但对 Puma 来讲也是毫无压力的，最终的瓶颈都不是在 Puma 上，而是在目标数据库的写入性能上。

高并发系统的设计思路

首先，这部分的代码能够在这里找到：传送门

完整的代码还包含了老数据被覆盖无数据可读时的数据源切换逻辑，还有当无消费者时关闭 RingBuffer 的逻辑。上面的代码已经被简化了不少，想看完整代码的话能够在上面的连接中看到。

之后有空还会再介绍更多 Puma 中遇到的问题和解决的思路。

而后谈谈高并发系统的设计。

Java 并发编程的第一重境界是善用各类锁，尽可能减小锁冲突，不能有死锁。

第二重境界就是善用 Java 的各类并发包，Java 的并发包里有的是无锁的，例如AtomicLong中用了CAS；有的是用了各类手段减小锁冲突，例如ConcurrentHashMap中就用了锁分段技术。总体效率都很是高，能熟练应用后也能写出很高效的程序。

再下一个境界就很是搞脑子了，每每是放弃了强一致性，而去追求最终一致性。其中会用到AtomicLong等无锁，或锁分段技术，而且经常会把它们结合起来用。就像上面那部分代码，看似简单，但实际上却要把各类边界条件思考地很全面，由于是最终一致性，因此中间的状态很是多。

源地址：http://www.dozer.cc/2016/09/ringbuffer-in-puma.html