前言

在介绍kafka的设计之前，先考大家一个问题，你知道怎么安全地用python写文件吗?

常见的写法：

with open("data.out", "w") as wf:
    wf.write("123")

调皮的写法：

for i in range(100):
    with open("data.out", "w") as wf:
        wf.write("123")

事务级别的写法：

import os
with open("data.out", "w") as wf:
    wf.write("123")
    wf.flush()
    os.fsync(wf.fileno())

这里 wf.flush 做的事情是把python内部的缓存强行刷新到系统缓存中，os.fsync 是把系统缓存强行刷到磁盘中。通过这种方式就完成了一种“事务级别”的写文件。因为事务需要保证ACID的特性（原子性、一致性、隔离性、持久性），其中的持久性（Durability）就需要确保数据已经写到磁盘。这样即使出现停机或者断电数据都不会丢失。

所以你是不是以为kafka就是使用第三种方式存储的呢？其实并不是，kakfa也是用的第一种方式的，在后文会详细说明为什么，现在我们来进入正文

Kafka是什么？

Kafka是一个分布式的消息中间件，以极少的资源消耗加上极强的处理性能著称。但是当需要进行读写数据时，kafka 仅仅做的就是读文件和写文件。而大家都知道磁盘的速度很慢，那么为什么kakfa可以做到这么快呢？

（一）"无缓存"

首先我们来看下下图，来自ACM Queue上的一篇关于大数据的文章。这个结果是作者分别测试了在顺序读和随机读下，机械磁盘、SSD和内存的速度差异。这个实验的数据是作者生成了一个 $10^9$ 的 int 数组（4字节，总共4GB）。

从图中我们可以看到无论是机械磁盘还是SSD，随机读的速度都很慢，分别只有1.2KB/s（316 * 4B）和 7.5KB/s（1924 * 4B）和对应的顺序读的速度差了4-5个数量级。但是我们可以看到磁盘的顺序读和内存的随机读的速度竟然是差不多，这个可能超出了我们大部分人的预料。

在大部分人的观念中磁盘的速度非常慢，记得读书时老师有个简单的比喻：CPU缓存的速度比内存快了3个数量级，内存的速度比磁盘快了3个数量级。但事实上这个说法只对了一半。因为日常应用中大部分都是随机写入，所以大家才会有这样一个观念：磁盘的速度比内存慢了三个数量级。因此读到这里你需要有一个新的观念：磁盘顺序读和内存随机读的速度是一个数量级的，甚至磁盘的速度更快。

将随机写变成顺序写

那么kafka是怎么把随机写入变成顺序写入呢？所以接下来我们需要说到操作系统了。因为磁盘随机读的软肋，现代操作系统会有一个非常激进的做法——在有多余内存的情况会愿意把文件都缓存到内存中。这样的好处是文件读写操作其实是在内存中处理的，速度非常快。而且因为磁盘批量写入的速度非常快，即使操作系统又需要回收这部分内存，也能快速得把这部分文件数据写到SWAP（虚拟内存，本质是磁盘）中。

除此以外，操作系统还会做一个“read-ahead”和“write-behind”的操作。也就是读文件的时候会做预测，提前多读一部分内容，在写的时候会把写的数据缓存起来，合并为一次批量的物理写入。

2. 为什么不用缓存？

接下来有同学可能会问了，就上述资料来说内存还是比磁盘快了7倍，为什么不将缓存直接维护在程序的内存中呢？这样的好处是可以不用依赖操作系统的缓存了，同时在程序运行时获得一个稳定的最优速度。在某些应用中这确实是个很好的方法，但在kafka中这样的收益并不高，会有以下几个考虑。

第一点，当你在代码中维护一份缓存的时候操作系统是并不知道你的行为的。所以它还是会默认的帮你缓存了一份文件在内存中。这时系统实际同时维护了两份文件的缓存，浪费了内存。内存不足时更容易触发内存和SWAP的切换：把一部分不使用的内存写入SWAP腾出内存空间，再从SWAP中找到需要的数据写到内存中。

第二点，进程自己维护一份缓存是不安全，一旦程序出现了crash对应的数据都会丢失。所以不可避免得进程需要定期把内容刷到磁盘中。而就像之前所说的，写到磁盘的操作其实是写到内存，等于把一次简单的内存写变成了两次内存写。

第三点，用程序维护一份缓存会出现冷启动的问题。因为软件运行初期需要把数据全部加载到内存中，数据量较大时可能会花费几分钟到几十分钟的时间。同时自己维护一份缓存需要更多的代码逻辑，而缓存常常是写代码最难的部分之一，把这个工作交给操作系统可靠又放心。

第四点，网络的限制。我们可以简单记住这个概念，大部分网络的传输速度是低于磁盘的，无论是随机写还是顺序写。所以根据木桶原理，没有必要达到这么快的速度。

综上，kafka 选择了这个直接“写硬盘”的设计思想，而这种内核编程思想是启发于一个高性能的反向代理服务器Varnish的设计思路。同样是一篇很棒的文章，Varnish作者在文章里面还提到了关于多核并行编程的小技巧，感兴趣可以了解一下。

（二）消息ACK机制

Kafka的消息ACK机制是我认为设计的第二精彩的地方。传统消息中间件都是服务端记录消费者消费的情况的，也就是具体消费到哪条数据了。简单看上去服务器自动记录这个数据是减轻了编写代码的难度，实际上并不是这样，特别是涉及到有数据修改的情况。

因为对于消息中间件来说，最难的问题是不知道消费者到底有没有成功接收并处理了这条数据。因为这条数据可能在传输过程中丢失了，可能消费者程序运行到一半的时候crash了。传统消息中间件想要确保消费成功就需要消费者再发送一个ACK给服务器，但是这样的缺点是造成性能的损耗。而且对于消费者来说，不同消息中间件采用的默认ACK策略是不同的，类似的隐藏规则会增加开发的难度。

所以kafka的选择的是让消费者自己维护这些数据。而消费者自己维护ACK的好处能让业务代码的逻辑非常清晰。就如《Zen of Python》中的第一句所说：“Beautiful is better than ugly.Explicit is better than implicit”。让业务自行决定ACK信息，就可以完全根据业务自身的特点来实现不同的逻辑。对于不重要的消息，可以选择收到消息就ACK，对于特别重要的消息可以等到完全消费成功再ACK，支持重复消费。

除此以外kafka是一个纯粹用硬盘的存储系统，这个关键因素决定了数据可以保留的时间非常长并且花销很小。这个特性加上自定义ACK的位置反过来也提供了一种新功能——重复消费以前信息的能力，一举两得。假设线上业务真的出现了一些bug也可以利用这些信息回溯以前全部的数据。

（三）数据传输

数据传输这部分也特别有意思，kafka做了一件事情：优化了字节复制的功能。

下面是一个传统socket服务端的简单示例，大家有个简单的印象。这个服务做的事情是把一个叫"foo.txt"的文件发送给客户端。

import socket
import os
import sys

CHUNK_SIZE = 8 * 1024

def start_server_traditional():
    """ 传统的方式 """
    server_socket = socket.socket()
    server_socket.bind(('localhost', 12346))
    server_socket.listen(5)
    while True:
        client_socket, addr = server_socket.accept()
        with open('foo.txt', 'rb') as f:
            data = f.read(CHUNK_SIZE)
            while data:
                client_socket.send(data)
                data = f.read(CHUNK_SIZE)
        client_socket.close()

start_server_traditional()

现在我们尝试从系统层面来分析一下这个过程：

1. The operating system reads data from the disk into pagecache in kernel space
2. The application reads the data from kernel space into a user-space buffer
3. The application writes the data back into kernel space into a socket buffer
4. The operating system copies the data from the socket buffer to the NIC buffer where it is sent over the network
1. 操作系统把数据从磁盘复制内核空间的缓存中。
2. 应用程序把内核空间的数据复制到用户空间的缓存。
3. 应用程序再次把数据复制到内核空间的socket缓存中。
4. 操作系统再把socket缓存中的数据复制到网卡中发送给网络。

可以看到示例代码的操作非常不高效，有四个复制操作和两个系统调用。而事实上unix操作系统有一个高度优化的方式可以直接把页缓存中的数据发送给socket，叫做sendfile system call。避免了多次不必要的复制，直接将数据发送到NIC中。

很好奇到底是怎么回事，所以自己用python做了一个小测试。实测下来效果很好，通过这个优化性能提升了3倍。以下是完整的测试代码，start_server_traditional是用传统的方式读文件并且发送，start_server_use_sendfile是用系统优化过的sendfile的方式。

运行方式：

python3 server.py 0 &
python3 server.py 1 &
python3 client.py

server端代码（server.py）

import socket
import os
import sys

CHUNK_SIZE = 8 * 1024

def start_server_traditional():
    """ 传统的方式 """
    server_socket = socket.socket()
   server_socket.bind(('localhost', 12346))
   server_socket.listen(5)
   while True:
        client_socket, addr = server_socket.accept()
        with open('foo.txt', 'rb') as f:
            data = f.read(CHUNK_SIZE)
            while data:
                client_socket.send(data)
                data = f.read(CHUNK_SIZE)
        client_socket.close()

def start_server_use_send_file():
    """ 优化的方式 """
    server_socket = socket.socket()
    server_socket.bind(('localhost', 12347))
    server_socket.listen(5)
    while True:
        client_socket, addr = server_socket.accept()
        with open('foo.txt', 'rb') as f:
            os.sendfile(client_socket.fileno(), f.fileno(), 0, 0)
        client_socket.close()


if __name__ == "__main__":
    mode = sys.argv[1]
    if mode == "0":
        start_server_traditional()
    elif mode == "1":
        start_server_use_send_file()

client代码(client.py)：

import socket
import time

CHUNK_SIZE = 8 * 1024

def run(port):
    sock = socket.socket()
    sock.connect(('localhost', port))
    chunk = sock.recv(CHUNK_SIZE)
    while chunk:
        chunk = sock.recv(CHUNK_SIZE)
    sock.close()

def test(port, n):
    t0 = time.time()
    for i in range(n):
        run(port)
    print("run port: %s, time_used:%.2fms" % (port, (time.time()-t0)*1000))

if __name__ == "__main__":
    n = 100
    test(12346, n)
    test(12347, n)
    test(12346, n)
    test(12347, n)

运行结果：

run port: 12346, time_used:1211.88ms
run port: 12347, time_used:413.35ms
run port: 12346, time_used:1226.00ms
run port: 12347, time_used:407.83ms

我们可以看到实现的性能非常的好，但这样也有一个缺点。服务端不能对数据做改动，因为一旦有改动又会退化为原先的较慢的模式。所以这也可以解释了kafka的另一个设计：kakfa的很多操作都是放在生产者和消费者做的，目的就是尽可能减少服务端的操作。比如生产者，服务端，消费者都使用了统一的标准二进制消息格式，CRC检验都是生产者实现的等

（四）容灾

传统的做法

容灾的原理是创建多个备份，当leader出现的故障时使用从库来代替主库完成相应的工作。

所以出现故障后首先会有新的leader选举操作，一个常见的解决方案是采用投票的方式——少数服从多数。这样的做法下，能支持n个单点故障的系统就会需要2n+1个备份，因为假设n个单点的数据出现了错误，至少有n +1 个节点的数据是正确的，那么投票结果一定是正确的结果。

根据这个说法，leader在确认收到消息之前只需要确保至少 n+1 份数据已经备份成功即可。因为当leader挂了，会需要选举谁是leader。那么按照投票的原则，n+1 份正确的数据一定占大多数的。

这个做法有一个好处，在 2n +1 个备份的情况下，只需要 n + 1 个从库ACK即可，所以这里leader ACK的速度其实是取决于最快的 n + 1 个从库。但这样的缺点是浪费机器资源，比如为了支持单点故障就需要3份数据，为了支持2个单点故障需要5份数据。而实际业务都需要达到2个单点故障的标准，也就是要5个机器。所以在大数据下这样的代价太高了。

kafka的优化

kafka在这个基础上做了一个的优化，维护了一个同步中机器的集合（ISR in-sync replicas）。在投票时只允许这个集合内的机器来投票，而不是采用全集。通过这个优化，kafka只需要ISR集合中的机器ACK即可。这样的实现下，n 点故障只需要 n+1 个机器。比如单点故障只需要两台机器，2个单点故障需要3台机器即可。不过这样的缺点也比较明显，因为需要n + 1个机器都ACK，leader的ACK速度取决于最慢的那个机器了。

但kakfa认为大多数场景下这个取舍是值得的，因为浪费2倍机器的代价太大了。而且kafka针对这个问题提供了一个优化方案。用户可以在写入的时候自定义最少的ACK数，假设系统设计考虑的是2个单点故障，也就是需要3个ACK，用户在考虑性能的情况下可以指定允许单个ACK即可，来提升性能。

（五）磁盘flush策略

现在让我们回到前言中的那个例子，为什么kafka选择了事务不安全的写法呢？

常见写法：

with open("data.out", "w") as wf:
    wf.write("123")

事务级别的写法：

import os
with open("data.out", "w") as wf:
    wf.write("123")
    wf.flush()
    os.fsync(wf.fileno())

首先第一点就是性能问题，fsync 这个系统调用会导致大量的小数据写。这样就不能充分利用操作系统对小数据写的优化：把随机写合并为顺序写。

第二点是因为"持久化"是基于磁盘不会出现问题的假设。而事实上生产服务器中磁盘出现问题才是最常见的情况。为了解决持久化更好的做法是使用多个机器来做灾备。

第三点是kafka不需要每个从库都保证数据的完整性。因为假设出现断电的情况就会造成文件写了一半的现象，也就是常说的文件损坏。但kafka能处理这种文件损坏的问题，因为它会对数据进行CRC校验，丢弃错误的数据并从其他机器同步。

基于以上的原因kakfa才敢选择了这个“大胆”的策略。不过这个fsync的操作还是很有意思的，特意做了一些简单的实验。测试下来性能差异在1.1倍到3倍之间，具体的影响因素和数据量大小还有硬盘类型有关。

以下是一个简单的实例代码，有兴趣的同学可以试着玩一下。

import os
import time

def test(file_flush=False, os_flush=False, file_path="data/foo.txt", to_write_str=""):
    foo_file = open(file_path, "w")
    foo_file.write(to_write_str)
    if file_flush:
        foo_file.flush()
    if os_flush:
        os.fsync(foo_file.fileno())
    foo_file.close()

def do(n, os_flush, to_write_str):
    t0 = time.time()
    for i in range(n):
        test(os_flush=os_flush, file_flush=True, to_write_str=to_write_str)
    print("os_flush:%s time_used:%.2fms" % (os_flush, (time.time()-t0)*1000))

if __name__ == "__main__":
    to_write_str = "This is test. This is test. This is test." * 100
    n = 1000
    do(n, os_flush=False, to_write_str=to_write_str)
    do(n, os_flush=True, to_write_str=to_write_str)
    do(n, os_flush=False, to_write_str=to_write_str)
    do(n, os_flush=True, to_write_str=to_write_str)

测试结果：

os_flush:False time_used:171.68ms
os_flush:True time_used:486.87ms
os_flush:False time_used:262.89ms
os_flush:True time_used:421.06ms

参考资料：

Apache Kafkakafka.apache.org

如何超过大多数人 | | 酷壳 - CoolShellcoolshell.cn

File transfer over sockets without user-space memory in Python 3.5michaldul.com

https://blog.csdn.net/ichuzhen/article/details/52207302blog.csdn.net

ITPub博客

从kafka设计中学到的开发技巧

前言