这个冠军的方案确实赞，10G的mysql binlog重放并传输只用了2秒！
总决赛冠军队伍 作死小分队 比赛攻略
决赛答辩PPT已上传！点这里查看。
赛题分析
给定一批固定的增量数据变更信息（存放在Server端），程序需要单线程顺序读取文件，进行数据重放计算，然后将最终结果输出到给定的目标文件（在Client端）中。
增量数据的变更主要包含数据库的Insert/Update/Delete三种类型的数据。主键可能发生变更。
为了降低数据在网络中的传输开销，我们的设计是在Server端完成数据的重放计算，再将结果发送到Client端，写入结果文件中。
比赛采用16核机器运行程序。显然，我们需要设计一个并行算法，充分利用多核CPU。
解题思路
由于数据的重放计算在Server端，Client主要负责接收结果和写入文件，因此核心的算法都在Server端。本节将首先给出Server端算法的系统架构，之后具体介绍多线程算法的细节与实现。
系统架构
为了最大限度利用多核CPU的计算能力，我们将整个重放过程按照流水线的方式分成三个部分：
Fig.1 系统架构图
1) Reader 按照赛题要求，负责单线程读取文件。事实上，由于单线程的内存拷贝速度已经跟不上流水线速度，我们用 MappedByteBuffer 将文件按照16M的大小切分成多个的分段（Segment）映射到内存，交由Parser处理。
为防止最后的一个 Task 被从中间切成两段，Segment 除了 16MB 数据以外还多保留了 128KB 的 Margin。
2) Parser 负责从 Segment 中初步 Parse 出 Task的操作类型 (Insert/Update/Delete) 以及主键值。根据主键值，将 Task 分成 N 个 Bucket 分别交给 Worker 处理。 Parser支持多线程执行，可伸缩。
如果是 UpdatePK, 则产生 UpdatePKSrc 和 UpdatePKDst 两个 task 以及对应的 Promise 对象，按照各自的 PK , 分别发给对应的 Bucket (Promise对象及Bucket之间如何协作处理主键更新问题下文会详细说明)。
Parser 产生的 task 不是一个一个发给 Worker/Bucket 的。这样吞吐量上不去。更好的方案是按批发送。如图所示，每个 Segment 对于每个 Bucket 产生一批 tasks，这些 tasks 被附加到对应的 Segment 上，供 Worker 读取。
容易误解的一点：Parser 并不是 Parse 全部的 key-value，这不够高效。Parser 只负责第一列也就是主键，剩下的部分，通过把当前的 offset 传给 Worker，从而交给 Worker 来处理。
3) Worker 和 Bucket 是一一对应关系，Worker 根据 Parser 产生的结果，在自己的 Bucket 上依次重放 task，多线程执行，可伸缩。
当所有 Worker 处理完最后的 task 时，意味着回放完成，可以准备输出了。输出其实是 merge K 个有序 stream 的经典问题，可以用堆来高效的解决。
上述的流水线非常适合用 RingBuffer 实现，原因如下：
* RingBuffer 相比 BlockingQueue 速度更快，且本身不产生新对象，减少 GC
* RingBuffer 能够方便地为 slot 静态分配内存空间
这里我们用了 Disruptor 框架，它是一个高性能的线程间消息通讯的库，底层用 RingBuffer 实现。它不仅能取代 ArrayBlockingQueue，功能上还要丰富的多。对于本题的架构，只需要一个 RingBuffer 就能完成。
以下的示意图是和 Fig.1 是 完全等价 的：
Fig.2 RingBuffer流水线效果图
图中白色部分为当前RingBuffer内的数据。图中 Worker 2 和 Worker 0 都在处理 Segment 103 中对应自己 Bucket 的 task。
并行算法
为了发挥并行性能，在每个Parser中我们将数据表按 hash(PK) 切分成 N 个 Bucket, 每个Bucket都由一个独立的Worker线程完成重放计算。
Fig.3 Hash 分桶示意图
对于Insert，Delete 和一般的 Update 事件，只要分配到对应的 Bucket 去做就可以了；唯独 UpdatePK（更新主键）事件例外，必须要 Bucket 间协作才能保证正确地把数据移动过去。
何谓“Bucket 间协作”？
一行数据被“拿走”的时候，如果还存在对该行的操作没完成，那这些修改就丢失了！所以，必须要保证一行数据被“拿走”前，所有的修改都已经 apply 到上面。反之同理，必须要“拿到”数据以后，才能把后续的操作 apply 上去。
也就是说，我们要让这两个线程在这一点（UpdatePK 这条操作）上同步才行。
Fig.4 UpdatePK 对 Bucket 的影响，接收方要等待发送方也处理到这个task
很自然的想到，可以用 CountDownLatch 来阻塞 UpdatePK 的接收方（数据被移到此 Bucket），直到 UpdatePK 的发送方发出这行数据，它才拿到数据、接着运行。然而，当 UpdatePK 操作较为密集的时候，这个解决方案非常低效！
另一种思路是内存中维护一张主键变更表，记录主键的变更历史，将所有 UpdatePK 后新主键的所有操作都分配到旧主键所在的Bucket中。然而每个task分配时都需要从主键变更表中查找对应的Bucket，并且Parser也无法并行执行，同样十分低效。
有什么更高效的解决方法呢?
针对这一问题, 最终设计出下文的并行算法, 本算法的核心在于通过 Promise 对象，解决了 Update 主键这一难题，从而使得数据表的各个 Bucket 的线程能够无锁、高效地协作。
所谓 Promise，是借鉴 Future/Promise 异步编程范式设计的一个数据结构。实现很简单，只要封装一个 volatile 的变量，如下所示（实际代码实现更复杂，仅为示例）：
final class Promise {
volatile T data;
public boolean ready() { return data != null; }
public T get() { return data; }
public void set(T data) { this.data = data; }
}
Promise 在 Parser parse 到 UpdatePK 事件时产生，发送方和接收方都持有它的引用：发送方获得 UpdatePKSrc 任务，只能写入Promise（set）；接收方获得 UpdatePKDst 任务，只能读取Promise（get）、以及检查数据是否 ready。通过 Promise 作为中间媒介，被操作的数据记录 data 就能从源 Bucket：hash(srcKey) 移动到目标 Bucket：hash(dstKey)。
Fig.5 Promise与发送方和接收方的关系
相比上一小节提到的 Latch 解决方案，Promise 不是阻塞接收方，而是告诉他：你要的数据还没准备好。明智的接收方将会先“搁置”这个消息，并把后来遇到的所有对这个 Key 的操作都暂存起来（放在blockedTable中，如图所示）。一旦某一时刻 Promise.ready() 为真，就可以把这个 data 放到对应的 Key 上了！暂存的操作也可以那时候再做。
如何从 blockedTable 中高效地找到 promise.ready() 的 task？事实上对于每个发送者（也就是其他bucket），我们只要检查最早阻塞的那个 task 是否 ready 就可以了。
Fig.6 Task 处理效果图，暂存被阻塞的Task
对于被 block 的 PK 的操作，会以一个链表保存下来。如果不巧操作很多，这个链表就会变的很长。一个简单的改进：如果来的是一个普通 Update 操作，其实可以直接 apply 到上一个操作上的 data 上，例如 A=5 B=6 可以叠加到 PK=3 A=4上， 就变成 PK=3 A=5 B=6，从而避免把 Update 操作追加到链表上。
附上关键逻辑的伪代码：
// 当 Parse 到 UpdatePK (SET PK=dstKey WHERE PK=srcKey)
Promise promise = new Promise();
bucket[hash(srcKey)].send(new UpdatePKSrcTask(srcKey, promise, ...));
bucket[hash(dstKey)].send(new UpdatePKDstTask(dstKey, promise, ...));
// 当接收方 Bucket 收到 UpdatePKDstTask task
if (!task.promise.ready()) {
LinkedList<Task> blockedTasksOnThisKey = new LinkedList<>(); // 存放阻塞的Task
blockedTasksOnThisKey.add(task);
blockedTable.put(task.key, blockedTasksOnThisKey); // 暂存，以后再处理
} else {
table.put(task.key, task.getData()); // 直接处理
}
// 当发送方 Bucket 收到 UpdatePKSrcTask task
task.promise.set(table.remove(key));
// 当接受方发现 blockedTable 中的 task.promise 已经 ready，则取出来处理掉
for (Task task : blockedTasks) {
applyUnblockedTask(task);
}
如果阻塞的 Tasks 中包含一个 Delete，后面又来了一个 UpdatePKDst，要注意，可能会再次阻塞。
您可能担心 blockedTable 的查询增加了单个 Bucket 的计算负担。实验表明，由于各个 Bucket 的工作进度差异相差不会很大，blockedTable 的最大 size 也在 25000 以内，远小于数据表大小，所以这个代价是完全可接受的。
健壮性分析
算法的正确性是最重要的。对于任意的输入数据集，算法都能保证输出正确的结果。
可以从理论上证明：本算法可以处理以任何顺序出现的 UpdatePK / Update / Delete / Insert 操作，保证重放结束后一定查询到正确的结果。
其实很简单，算法保证了所有的操作都在它们可以执行的时候被执行，换句话说，对于一切有互相依赖关系的操作，算法不会破坏它们的先后关系。算法的并行性，是在保证了该前提的情况下做到的。
以下的例子可能帮助您获得一个感性的认识。
从一个简单的例子开始。如果遇到如下的序列
Insert PK=1 A=1 B=2 Update A=2 Where PK=1 Update PK=2 Where PK=1 Update A=3 Where PK=2假设 PK = 1 和 2 分别被 hash 到 Bucket 1 和 Bucket 2，那么会有如下情况：
Insert PK=1 A=1 B=2 // Bucket 1 新增记录 Update A=2 Where PK=1 // Bucket 1 更新记录 Update PK=2 Where PK=1 // Bucket 1 接到 UpdatePKSrc，移除并把该数据 set 到 promise // Bucket 2 如果拿到了 data，那就成功了；假设它没拿到，PK=2加入阻塞表 Update A=3 Where PK=2 // Bucket 2 把这条 Update 追加到 PK=2 的操作列表上 ... // Bucket 2 等到了 data，Update A=3 也被重放了举一个更极端的例子来说明。如果遇到这样的情况：
Insert PK=1 A=1 B=2 Update A=2 Where PK=1 Update PK=2 Where PK=1 Update PK=3 Where PK=2 // 连续的更新 Update A=3 Where PK=3算法会做如下处理：
Insert PK=1 A=1 B=2 // Bucket 1 新增记录 Update A=2 Where PK=1 // Bucket 1 更新记录 Update PK=2 Where PK=1 // Bucket 1 接到 UpdatePKSrc，移除并把该数据 set 到 promise // Bucket 2 如果拿到了 data，那就成功了；假设它没拿到，PK=2加入阻塞表 Update PK=3 Where PK=2 // Bucket 2 发现 2 这个主键在阻塞表中，所以本操作也放入阻塞表 // Bucket 3 无法拿到 data，所以把 PK=3 加入阻塞表 Update A=3 Where PK=3 // Bucket 3 把这条 Update 追加到 PK=3 的操作列表上 ... // Bucket 2 等到了 data，UpdatePK 也被重放了 ... // Bucket 3 等到了 data，Update A=3 也被重放了不妨自己尝试更多的情况。
关于表结构的健壮性, 程序会根据第一次遇到的 Insert log 来确定表结构，包括各个列的名字、类型、主键信息等。
程序严格按照比赛要求。对于数字，支持 long 型正数范围；对于文本，最长支持 65536 个字符。具体实现参考下文“数据存储”一小节。
优势与创新点
在健壮性的基础上，本算法还有以下几点优势：
* 完全无锁（Lock-free），无阻塞（Non-blocking）。在16核CPU的测试场景下，锁竞争将会导致不小的开销；而阻塞更不用说，极端情况下可能多线程会退化成协程。（例如 Latch 的解决方案，连续 UpdatePK 就会导致这样的情况）。本算法完全摈弃了wait()或lock()，而是用 代价极低 的 volatile 实现同步，这是最大的创新点。
* 可伸缩（Scalability）。除了 Reader 根据题意必须单线程，算法中没有任何不可伸缩的数据结构，理论上为线性加速比。若 CPU 核数增加，只要提升 Parser 和 Worker/Bucket 的线程数即可。（一些解决方案用到全局的 KeyMap，导致无法伸缩）
* 流处理（Streaming）。本算法是一个真正的流处理系统，在真实场景中可以不断灌入新数据并提供查询（保证最终一致性）。这也与比赛的初衷一致。
细节实现和优化
上述算法和架构给出了大致的代码编写思路。细节上，为了追求极致的性能，我们还做了各种优化。
原生类型数据结构
Java 的范型对于 primitive type 的数据是严重的浪费。比如 Map 是非常低效的，不仅浪费了大量内存，还产生了大量冗余的 boxing/unboxing。
对此，我们利用 fastutil 和 koloboke 这两个库代替 Java 标准库中范型实现的 HashMap、ArrayList 等数据结构，极大提升了性能。
数据存储
我们使用 long 数组来存储每一行数据。
由于列值的类型为 long 或者 String。对于 long 类型值, 将其解析成 long 数据存储即可。而对于String 类型的数据, 如果通过将其转换成 String 存储, 至少有两个问题：
- Encode/decode 造成无谓的性能损耗；
- 内存开销很大，对象数量非常多，对 GC 造成巨大压力。
StringStore 类能够将 String 类型的列也“变成” long，从而放到 long 数组中。利用中间结果文件，创建一个 MappedByteBuffer, 将字符串 bytes 写入 MappedByteBuffer 中，并将 position 和 length 用位运算压缩到一个 long 值中返回。根据这个值即可从 MappedByteBuffer中读取出字符串数据。
一个优化是：如果字符串的 bytes 的长度小于等于 7: 那么直接利用 long 里面的 7 个字节存储，剩下一个字节存长度，避免了磁盘写入。附上StringStore类的核心代码：
// 写入byte[]类型的数据，范围long值作为读取的索引
public long put(byte[] data, int len) {
if (len <= 7) {
long value = unsafe.getLong(data, BYTE_ARRAY_BASE_OFFSET);
return uint64(len) << 56 | (value & 0xffffffffffffffL);
} else {
long pos = doPut(data, len);
return 0xffL << 56 | pos << 32 | len;
}
}
// 根据写入时获得的long值，读取相应的数据
// 为了减少内存中的对象拷贝，直接将结果写入ByteBuffer中
public static void get(long value, ByteBuffer buf) {
int h = (int) (value >>> 56);
if (h != 0xff) {
buf.putLong(value);
buf.position(buf.position() - 8 + h);
} else {
int pos = (int) ((value & 0xffffff00000000L) >>> 32);
int len = (int) (value & 0xffffffffL);
doGet(pos, len, buf);
}
}
数组池
如果为每行数据都创建一个 long[]，需要频繁地 new 出大量对象。为此，我们实现了一个 LongArrayPool 来管理所有的行数据，用 offset 来查找所需的数据。
对象池 + 数组化
以 Segment 中附加的 tasks 为例，如果每次 new Task() 将产生总计近亿个 Task 对象，造成严重的 GC 压力，显然不可取。
使用对象池可以解决一半的问题。通过复用 Task，减轻了 new Task() 的压力。但这还不够好！让我们看看 Task 的结构：
final class Task {
byte opcode;
long key;
int promise;
int data;
}
可见，Task 本身结构很简单，相比之下对象头的代价显得很浪费。如果不用对象，其实可以用数组来代替：
// In Segment.ensureAllocated()
opcodes = new byte[Constants.SEGMENT_MAX_EVENTS_NUM_PER_BUCKET];
offsets = new int[Constants.SEGMENT_MAX_EVENTS_NUM_PER_BUCKET];
keys = new long[Constants.SEGMENT_MAX_EVENTS_NUM_PER_BUCKET];
promises = new int[Constants.SEGMENT_MAX_EVENTS_NUM_PER_BUCKET];
这样做有以下几个优点：
- 创建（分配内存）速度大大提升
- 内存占用大幅下降，省掉了对象头的开销
- 确保连续的内存分配，顺序访问更快
对于 Promise 可以做类似的优化，参考 PromisePool，这里不再赘述。
Pool 的懒初始化
做了上述数组池和对象池优化后，程序启动时间大大增加，这是因为创建 Pool 需要大量分配内存，如果发生在类加载期间，就会阻塞 main 函数的运行。
解决方案是适当延迟部分 Pool 的分配，对它们采用 lazy 的初始化策略，即第一次使用时才分配所需的内存空间。
GC 调优
JVM 会在新生代不够分配时触发 GC。考虑到我们有 1G 的新生代内存，而事实上要动态 new 的对象很少，通过调节 Pool 的初始化时机，可以做到只发生一次 ParNew GC。
对于老年代的 CMS GC 代价很大，我们在比赛中尽可能避免触发 CMS GC。而这就要求尽可能节约内存，上文提到的对象池和数组池发挥了重要作用。
线程数调优
根据比赛数据集选取最合适的 Parser 和 Worker 线程数，对榨干最后一点 CPU 性能至关重要。
为了调优, 我们启动一个 monitor 的 daemon 线程, 定时打印 Reader, Parser 和 Worker的进度，从而推测性能瓶颈在哪一方。经尝试，我们将 Parser 线程定为 6 个，Worker（即 Bucket 数）定为 10 个。
Parser 读取预测
由于题目规定为单库单表，"变更类型"（U/I/D）之前的字符（下称 header）没有必要解析，可直接跳过。不过这部分的字符长度并不确定, 因此我们尝试预测这个跳过的长度。对于每一行，如果推测正确，则可以直接跳过这部分字符；否则，从行首开始解析直到到达"变更类型", 同时更新预估的 header 长度。
由于大部分的 log 的 header 长度是一样的, 这个技巧有效地避免大量不必要字符的解析。
同理，对于 Parser 来说除了主键以外剩下 key-value 并没有用。用类似的思路也可以预测长度并直接跳过。附上 skipHeader的代码：
if (buffer[pos + opcodeOffset - 1] == (byte) '|'
&& buffer[pos + opcodeOffset + 1] == (byte) '|') {
// Fast pass
pos += opcodeOffset;
} else {
int lineBegin = pos;
pos++; // Skip '|'
pos = skipNext(buffer, pos, '|'); // Skip binlog ID
pos = skipNext(buffer, pos, '|'); // Skip timestamp
pos = skipNext(buffer, pos, '|'); // Skip database name
pos = skipNext(buffer, pos, '|'); // Skip table name
opcodeOffset = pos - lineBegin;
}
重写网络传输和 Logging
实验发现 Netty 和 logback 都比较重量级，拖慢了启动速度。因此自己实现了网络传输和 Logger，减少启动时间。
总结与感想
算法决定了性能的上限，工程实现的好坏决定了能多大程度接近这个上限。
大赛过程竞争非常激烈，可谓高手云集。通过这场比赛，我们的技术得到了锻炼，收获了解决问题的成就感。同时也真诚感谢大赛的主办方，让我们有机会在赛场上证明自己的能力。
