Kafka 设计详解之队列

   <h2><strong>前言</strong></h2>    <p>本文打算详细分析 Kafka 的核心 — <strong>队列</strong> 的设计和实现，来对 Kafka 进行更深一步的了解。</p>    <h2><strong>如何设计队列</strong></h2>    <p>队列是一种先进先出（FIFO）的数据结构，它是 Kafka 中最重要的部分，负责收集生产者生产的消息，并将这些消息传递给消费者。要实现一个队列有多种方式，Kafka 作为一个消息队列中间件，在设计队列时主要要考虑两个问题：</p>    <p><strong>1. 队列数据是写到内存还是写到磁盘</strong></p>    <p>乍一看到这个问题，我们会想，内存的读取速度远快于磁盘，如果追求性能，内存也充足的话，当然是将生产者产生的消息数据写到内存（比如用一个数组或者链表来存储队列数据），供消费者消费。真的是这样吗？</p>    <p>下面我们依次分析下写内存和写磁盘文件的优缺点，首先，内存的优点是读写速度非常快，但是，如果我们的目标是设计「大数据量」下的「高吞吐量」的消息队列，会有以下几个问题：</p>    <ul>     <li>GC 的时间消耗：如果队列数据都在内存，数据会非常大（几十 G）， 因为消息队列需要不断地接受新产生的消息和删除已经被消费的消息（不然内存很快会被撑爆），Java GC 的消耗不容忽视。</li>     <li>Java 内存存储效率：如我们所知，一个 Java 对象的内存开销会大于其对象数据本身，通常对象的内存开销是数据本身的一倍（甚至更多）。</li>     <li>设计复杂度增加：为保证数据不丢失，需要一个预写日志（WAL），如果程序异常挂掉，重启时可以从 WAL 恢复数据。</li>     <li>数据初始化的时间消耗：程序重启时需要从文件将数据 load 到内存，如果数据对象大小是 10G 级别的，也会消耗大量的时间（10 分钟左右）。</li>    </ul>    <p>接下来我们来分析一下磁盘，写磁盘文件方式存储队列数据的优点就是能规避上述内存的缺点，但其有很严重的缺点，就是读写速度慢，如果纯依靠磁盘，那消息队列肯定做不到「高吞吐量」这个目标。</p>    <p>分析了内存跟磁盘的优缺点，好像我们还是只能选写内存，但我们忽视了磁盘的两个情况：一是磁盘慢是慢在随机读写，如果是顺序读写，他的速度能达到 600MB/sec（RAID-5 磁盘阵列），并不慢，如果我们尽可能地将数据的读写设计成顺序的，可以大大提升性能。二是 <strong>现代的操作系统会（尽可能地）将磁盘里的文件进行缓存</strong> 。</p>    <p>有了操作系统级别的文件缓存，那用磁盘存储队列数据的方式就变得有优势了。首先，磁盘文件的数据会有文件缓存，所以不必担心随机读写的性能；其次，同样是使用内存，磁盘文件使用的是操作系统级别的内存，相比于在 Java 内存堆中存储队列，它没有 GC 问题，也没有 Java 对象的额外内存开销，更可以规避应用重启后的内存 load 数据耗时的问题，而且，文件缓存是操作系统提供的，因为我们只要简单的写磁盘文件，系统复杂性大大降低。</p>    <p>因此，Kafka 直接使用磁盘来存储消息队列的数据。</p>    <p><strong>2. 以何种数据结构存储队列数据</strong></p>    <p>刚才我们已经决定用磁盘文件来存储队列数据，那么要如何选择数据结构呢？一般情况下，如果需要查找数据并随机访问，我们会用 B+ 树来存储数据，但其时间复杂度是 O(log N)，由于我们设计的是消息队列，我们可以完全顺序的写收到的生产者消息，消费者消费时，只要记录下消费者当前消费的位置，往后消费就可以了，这样可以对文件尽可能的进行顺序读写，同时，时间复杂度是O(1)。其实，这跟我们写日志的方式很像，每条日志顺序 append 到日志文件。</p>    <h2><strong>队列实现</strong></h2>    <p>之前我们已经确定采用直接顺序写磁盘文件的方式来存储队列数据，下面我们来剖析下具体的实现细节。</p>    <p>在 Kafka 中，用一个文件夹存储一条消息队列，成为一个 Log，每条消息队列由多个文件组成，每个文件称为一个 LogSegment，每当一个 LogSegment 的大小到达阈值，系统就会重新生成一个 LogSegment；当旧的 LogSegment 过期需要清理时（虽然磁盘空间相对于内存会宽裕很多，我们可以保存更长时间的消息数据，比如一周，以供消费者更灵活的使用，但还是需要定期清理太老的数据），系统会根据清理策略删除这些文件。</p>    <p>现在我们知道一个队列（Log）是由多个队列段文件（LogSegment）组成的，那么 Kafka 是如何将这些文件逻辑上连接从而组成一条有序队列的呢？在生成每个队列段文件时，Kafka 用该段的初始位移来对其命名，如在新建一个队列时，会初始化第一个队列段文件，那么其文件名就是0，假设每个段的大小是固定值 L，那么第二个段文件名就是 L，第 N 个就是 （N - 1）* L。这样，我们就可以根据文件名对段文件进行排序，排序后的顺序就是整个队列的逻辑顺序。</p>    <h2><strong>队列读写</strong></h2>    <p>了解了队列的基本实现，下面我们就来分析下队列的核心操作—读和写。</p>    <h3><strong>写</strong></h3>    <p>写操作发生在生产者向队列生产消息时，在上篇文章讲网络通信时我们已经说到，所有的客户端请求会根据协议转到一个 Handler 来具体处理，负责写操作的 Handler 叫 ProducerHandler，整个写请求的流程如下：</p>    <p style="text-align: center;"><img src="https://simg.open-open.com/show/00bd7af8f33e47ea599accb9b3e15f9d.png"></p>    <p style="text-align: center;">写消息流程</p>    <ol>     <li> <p>根据消息的 topic、partition 信息（跟其分布式有关，以后文章会专门说到），定位到具体的队列，即上面我们说到的 Log。</p> </li>     <li> <p>检验生产者消息是否合法</p> </li>     <li> <p>我们知道一个 Log 由多个 LogSegment 组成，在任意时刻， <strong>只有最新一个 LogSegment 是可写的</strong> ，其他的都是只读的，所以，接下来我们通过排序后的 LogSegment 获取到最新一个可写的 LogSegment。</p> </li>     <li> <p>用 NIO，将消息（一个 ByteBuffer）写到 LogSegment 所属的文件</p> </li>     <li> <p>上一步中说的写文件，考虑到效率问题，并没有直接将消息 flush 到磁盘，所以，这里其实 <strong>存在一个丢消息的风险</strong> 。在本步骤中，主要检查待 flush 的消息大小是否到达指定阈值，如果到了，就 flush 到磁盘</p> </li>     <li> <p>检查最新的 LogSegment 大小是否到达阈值，如果是，则保存关闭当前文件，新建一个 LogSegment 文件。</p> </li>    </ol>    <p>之前我们说过，如果是顺序写，由于省掉了磁头寻址的时间，磁盘的性能还是很高的，我们看到 Kakfa 队列是以顺序方式写的，所以性能很高。但是，如果一台 Kafka 服务器有很多个队列，而硬盘的磁头是有限的，所以还是得在不同的队列直接来回切换寻址，性能会有所下降。</p>    <h3><strong>读</strong></h3>    <p>队列的读操作发送在消费者消费队列数据时，由于队列是线性的，只需要记录消费者上次消费到了哪里（offset），接下去消费就好了。那么首先会有一个问题，由谁来记消费者到底消费到哪里了？</p>    <p>一般情况下，我们会想到让服务端来记录各个消费者当前的消费位置，当消费者来拉数据，根据记录的消费位置和队列的当前位置，要么返回新的待消费数据，要么返回空。让服务端记录消费位置，当遇到网络异常时会有一些问题，比如服务端将消息发给消费者后，如果网络异常消费者没有收到消息，那么这条消息就被「跳过」了，当然我们可以借鉴二阶段提交的思想，服务端将消息发送给消费者后，标记状态为「已发送」，等消费者消费成功后，返回一个 ack 给服务端，服务端再将其标记为「成功消费」。不过这样设计还是会有一个问题，如果消费者没有返回 ack 给服务端，此时这条消息可能在已经被消费也可能还没被消费，服务端无从得知，只能根据人为策略跳过（可能会漏消息）或者重发（可能存在重复数据）。另一个问题是，如果有很多消费者，服务端需要记录每条消息的每个消费者的消费状态，这在大数据的场景下，非常消耗性能和内存。</p>    <p>Kafka 将每个消费者的消费状态记录在消费者本身（隔一段时间将最新消费状态同步到 zookeeper），每次消费者要拉数据，就给服务端传递一个 offset，告诉服务端从队列的哪个位置开始给我数据，以及一个参数 length，告诉服务端最多给我多大的数据（批量顺序读数据，更高性能），这样就能使服务端的设计复杂度大大降低。当然这解决不了一致性的问题，不过消费者可以根据自己程序特点，更灵活地处理事务。</p>    <p>下面就来分析整个读的流程：</p>    <p style="text-align: center;"><img src="https://simg.open-open.com/show/cf217c5000fc7c834e0f45b5bbfee78a.png"></p>    <p style="text-align: center;">读消息流程</p>    <ol>     <li> <p>在收到消费者读的请求后，根据请求中的 topic、partition 信息，定位到具体的队列（Log）。</p> </li>     <li> <p>根据 offset（我们要开始消费的队列位置），因为每个 LogSegment 文件都是以队列的位置命名的，所以可以利用 offset 进行二分查找寻找具体的 LogSegment。</p> </li>     <li> <p>在找到具体的 LogSegment 后，就可以读数据了，不过，在这里并不真正读数据，而是生成一个引用，记录该文件的 channel，这次要读取数据在文件中的起始位置以及结束位置，在真正进行网络传输时，我们利用 <strong>零拷贝（zero-copy）</strong> 将数据传输，即从文件的 channel 直接向 socket channel 传输数据（Java 中是 channel.transferTo() 方法）。</p> </li>     <li> <p>消费者收到返回的数据后，解码成真正的 message 列表。</p> </li>    </ol>    <h2><strong>一致性问题</strong></h2>    <p>分布式系统中不可避免的会遇到一致性问题，主要是两块：生产者与队列服务端之间的一致性问题、消费者与队列服务端之间的一致性问题，下面依次展开。</p>    <p>生产者与队列服务端之间的一致性</p>    <p>当生产者向服务端投递消息时，可能会由于网络或者其他问题失败，如果要保证一致性，需要生产者在失败后重试，不过重试又会导致消息重复的问题，一个解决方案是每个消息给一个唯一的 id，通过服务端的主动去重来避免重复消息的问题，不过这一机制目前 Kafka 还未实现。目前 Kafka 提供配置，供用户不同场景下选择允许漏消息（失败后不重试）还是允许重复消息（失败后重试）。</p>    <p>消费者与队列服务端之间的一致性</p>    <p>由于在消费者里我们可以自己控制消费位置，就可以更灵活的进行个性化设计。如果我们在拉取到消息后，先增加 offset，然后再进行消息的后续处理，如果在消息还未处理完消费者就挂掉，就会存在消息遗漏的问题；如果我们在拉取到消息后，先进行消息处理，处理成功后再增加 offset，那么如果消息处理一半消费者挂掉，会存在重复消息的问题。要做到完全一致，最好的办法是将 offset 的存储与消费者放一起，每消费一条数据就将 offset+1。</p>    <h2><strong>总结</strong></h2>    <p>本文介绍了 Kafka 的队列实现以及其读写过程。Kafka 认为操作系统级别的文件缓存比 Java 的堆内存更省空间和高效，如果生产者消费者之间比较「和谐」的话，大部分的读写操作都会落在文件缓存，且在顺序读写的情况下，硬盘的速度并不慢，因此选择直接写磁盘文件的方式存储队列。在队列的读写过程中，Kafka 尽可能地使用顺序读写，并使用零拷贝来优化性能。最后，Kafka 让消费者自己控制消费位置，提供了更加灵活的数据消费方式。</p>    <p> </p>    <p>来自：http://www.jianshu.com/p/6b2e39ba7787</p>    <p> </p>