本文共 5555 字，大约阅读时间需要 18 分钟。

1、什么是Kafka？

官方定义：Apache Kafka is an open-source distributed event streaming platform(Kafka是一个开源的分布式流计算平台)

自定义：Kafka是一个分布式、分区的、多副本的、多生产者、多订阅者、基于zookeeper协调的分布式日志系统和消息系统。

2、运用场景

主要是日志搜集系统和消息系统

3、消息系统

消息传递模式：点对点模式 和 发布-订阅模式

即：生产者发送一条消息到queue，只有一个消费者能收到。 当发布一个消息，所有订阅这个topic的服务都能得到这个消息，所以从1到N个订阅者都能得到同一个消息的拷贝。

Kafka：

消费端为拉模型，消费状态和订阅关系由客户端端负责维护，消息消费 完后不会立即删除，会保留历史消息。因此支持多订阅时，消息只会存储一 份就可以了。

4、Kafka-扮演角色

4.1 消息系统

Kafka和传统的消息系统（消息中间件）都具备系统解耦、冗余存储、流量削峰、缓冲、异步通信、扩展性、可恢复性等功能；另外，Kafka还实现了消息顺序性保障，回溯消费的功能。

4.2 存储系统

Kafka把消息持久化到磁盘，相比于其他基于内存存储的系统而言，有效地降低了数据丢失的风险。也正是得益于Kafka的消息持久化功能和多副本机制，我们可以把Kafka作为长期的数据存储系统来使用，只需要把对应的数据保留策略设置为”永久”或启用主题的日志压缩功能。

4.3 流式处理平台

Kafka不仅为流式处理框架提供了可靠的数据来源，还提供了一个完整的流式处理类库，如：窗口、连接、变换、聚合等各类操作。

5、Kafka-高吞吐低延迟

Kafka的数据虽然是写入磁盘，但写入速度是很快的，因为每次都是写入操作系统的页缓存（page cache）,然后由操作系统自行决定什么时候把页缓存中的数据写入磁盘上。

5.1 操作系统缓存

页缓存是内存分配的，Kafka不直接参与物理的I/O操作，而是交由操作系统来完成。

5.2 顺序读

Kafka采用追加(append)方式，避免了磁盘的随机读写操作（类比redis的aof持久化）。

5.3 零拷贝

使用sendfile为代表的零拷贝技术加强网络间的数据传输效率；数据直接在内核空间完成输入和输出，不需要拷贝到用户空间再写出去；kafka数据写入磁盘前，数据先写到进程的内存空间。

内核空间和用户空间：

那么为什么从内核空间转向用户空间这段过程会慢呢？首先我们需了解的是什么是内核空间和用户空间。在常用的操作系统中为了保护系统中的核心资源，于是将系统设计为四个区域，越往里权限越大，所以Ring0被称之为内核空间，用来访问一些关键性的资源。Ring3被称之为用户空间。 用户态、内核态：线程处于内核空间称之为内核态，线程处于用户空间属于用户态。

那么我们如果此时应用程序（应用程序是都属于用户态的）需要访问核心资源怎么办呢？那就需要调用内核中所暴露出的接口用以调用，称之为系统调用。例如此时我们应用程序需要访问磁盘上的文件。此时应用程序就会调用系统调用的接口 open方法，然后内核去访问磁盘中的文件，将文件内容返回给应用程序。大致的流程如下：

6、Kafka-基础架构

Kafka集群中按照主题(topic)分类管理一个主题可以有多个分区(partition)，一个分区可以有多个副本。 消息结构：键、值、时间戳。

7、Kafka-生产者

7.1 Producer

消息生产者，负责创建消息，将消息投递到Kafka中。

7.2 Topic

Kafka中的消息以主题的方式进行归类，生产者负责将消息发送到特定的主题中。

7.3 Partition

主题(Topic)是一个逻辑概念，它还可以细分为多个分区(Partition)。Topic与Partition是1：n的关系，即：一个分区只属于单个主题，同一个主题可能包含多个分区，不同分区包含的消息是不同的。

7.4 Interceptor

拦截器/拦截器链，用来过滤消息，修改消息，统计等。

7.5 Serializer(序列化器)

key.serializer

指定键的序列化器。Broker希望接收到的消息的键和值都是字节数组(默认)。这个属性必须被设置为一个实现了org.apache.kafka.common.serialization.Serializer接口的类，生产者会使用这个类把键对象序列化成字节数组。Kafka客户端默认提供了ByteArraySerializer、StringSerializer和IntegerSerializer，因此一般不需要实现自定义的序列化器。需要注意的是，key.serializer属性是必须设置的，即使只发送值内容。 value.serializer 指定值的序列化器。如果键和值都是字符串，可以使用与key.serializer一样的序列化器，否则需要使用不同的序列化器。

7.6 Partitioner(分区器)

默认DefaultPartitioner

如果key不为null，那么默认的分区器会对key进行哈希(采用MurmurHash2算法，具备高运算性能及低碰撞率)，最终根据得到的哈希值来计算分区号，拥有相同key的消息会被写入同一个分区；如果key为null，那么消息将会以轮询的方式发往主题内的各个可用的分区，可以自定义。

7.7 重要参数

acks

0：Producer 往集群发送数据不需要等到集群的返回，不确保消息发送成功。安全性最低但是效率最高。 1：Producer 往集群发送数据只要 Leader 成功写入消息就可以发送下一条，只确保 Leader 接收成功。 -1/all：roducer 往集群发送数据需要所有的 ISR Follower 都完成从 Leader 的同步才会发送下一条，确保Leader 发送成功和所有的副本都成功接收，安全性最高，但是效率最低。

bootstrap.servers

Kafka Broker集群地址

key.serializer

发送的Key序列化类

value.serializer

发送的value序列化类

max.request.size

生产者能发送消息的最大值，默认1MB

retries

生产者充实次数，默认是0

retry.backoff.ms

设置两次重试之间的时间间隔，避免无效的频繁重试，默认100ms

compression.type

消息压缩方式，默认值为”none”表示消息不会压缩，该参数还可以配置为“gzip”、“snappy”、“lz4”等。

batch.size

每个 ProducerBatch 要存放 batch.size 大小的数据后，才可以发送出去。比如说 batch.size 默认值是 16KB，那么里面凑够 16KB 的数据才会发送。

linger.ms

生产者客户端会在 ProducerBatch 填满或等待时间超过 linger.ms 值时发送出去。增大这个参数的值会增加消息的延迟，但是同时能提升一定的吞吐量。

buffer.memory

RecordAccumulator缓存的大小，默认值为 33554432B （32M）

max.block.ms

如果记录发送到RecordAccumulator的速度比sender发送到服务器的速度快，超过了RecordAccumulator设置的缓存大小，Producer 就会阻塞，如果阻塞的时间超过 max.block.ms 配置的时长，则会抛出一个异常，默认值为 60000ms

8、 Kafka-消费者

8.1 消费位移

offset：

消费位移存储在Kafka内部的主题(topic)_consumer_offsets中，消费者在消费完 消息之后需要提交offset；在Kafka中默认的消费位移是自动提交的，自动提交指的是 定期提交，而不是没消费一条就提交一条；而手动提交可细分为同步提交和异步提交。

enable.auto.commit：

是否自动提交消息位移，默认为true。

auto.commit.interval.ms：

自动提交消费位移时间间隔，默认是5s。

auto.offset.reset（该参数用于控制消息从哪里开始消费）

“earliest”：当分区下有已提交的offset时，从提交的offset开始消费；无提交的 offset时，从头开始消费。 “latest”：当分区下有已提交的offset时，从提交的offset开始消费；无提交的 offset时，消费新产生的该分区下的数据。 “none”：topic各分区都存在已提交的offset时，从offset后开始消费；只要有 一个分区不存在已提交的offset，则抛出异常。

8.2 消费线程

一般而言，分区是消费线程的最小划分单位；所以理论上来说，分区数和消费者数应该是1:1的；因为默认情况下一个分区只能被一个消费者消费，而一个消费者则可以消费多个分区；如果消费者个数大于分区个数的话，就会有消费者消费不到任何分区，处于空闲状态。

主流且推荐的方式：一个消费线程消费一个分区。 非主流方式：多个消费线程同时消费同一个分区，需要通过assign()、seek()等方法自定义实现，这样可以打破原有的消费线程的个数不能超过分区数的限制，进一步提高消费能力，不过这种实现方式对于位移提交和顺序控制的处理就会变得异常复杂，实际应用中使用极少

8.3 分区重分配

rebalance：

表示消费者组下的所有consumer通过某种规则来分配所订阅topic的每个 分区。所分配的这个过程称为rebalance；也可以说，将分区的所有权从一个消费者移 动到了另一个消费者称为rebalance。 什么时候触发rebalance：①、消费者组加入/减少消费者；②、消费者组订阅topic分 区发生变化；③、消费者组订阅topic数量发生变更。 如何进行rebalance：partition.assignment.strategy（分区分配策略） RangeAssignor（默认） Range策略是对每个主题而言的，首先对同一个主题里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。在我们的例子里面，排完序的分区将会是0, 1, 2, 3, 4, 5, 6, 7, 8, 9；消费者线程排完序将会是C1-0, C2-0, C2-1。然后将partitions的个数除于消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽，那么前面几个消费者线程将会多消费一个分区。在我们的例子里面，我们有10个分区，3个消费者线程， 10 / 3 = 3，而且除不尽，那么消费者线程 C1-0 将会多消费一个分区，所以最后分区分配的结果看起来是这样的： 假如我们有11个分区，那么最后分区分配的结果看起来是这样的： 假如我们有2个主题(T1和T2)，分别有10个分区，那么最后分区分配的结果看起来是这样的： RoundRobinAssignor：略 StickyAssignor：略 自定义分区策略：

8.4 重要参数

Bootstrap.servers：

Kafka Broker集群地址

group.id：

消费者组名称

client.id：

消费者客户端的id，如果不设置，KafkaConsumer会自动生成一个字符串 “consumer-”与数字的拼接。

key.deserializer和value.deserializer：

key和value对应的反序列化器。

fetch.min.bytes：

拉取数据时，一次拉取的最小数据量，默认1B。

fetch.max.bytes：

拉取数据时，一次拉取的最小数据量，默认50MB。

fetch.max.wait.ms：

Kafka等待时间，如果Kafka中没有足够多的消息而满足不了fetch.min.bytes参数的要求，那么等待的时间。

max.poll.records：

一次拉取请求中拉取的最大消息数，默认500条。

enable.auto.commit：

默认值是 true，采用自动提交的机制。

auto.commit.interval.ms：

默认值是 5000，单位是毫秒

9、 Kafka-小问题

9.1 Kafka能否保证消息有序？

Kafka的分区在存储层面可以看作是一个可追加的日志（Log）文件，消息在被追加到分区日志文件的时候都会有一个特定的偏移量(offset)，Kafka通过它来保证消息在分区内的顺序性，不过offset只针对分区，也就是说，Kafka保证的是分区(Partition)有序而不是主题(Topic)有序。因此，如果想要消息有序，可以使用单分区，单这也就相当于抛弃了Kafka的主要功能。

9.2 Kafka与MQ在消费上有啥区别？

Kafka消息消费后可以保留；消息集群可以选择保留所有消息，无论他们是否被消费，并通过一个可配置的参数（log.retention.hours、log.retention.minutes、log.retention.ms）来控制。

9.3 Kafka是否支持读写分离？为什么？

不支持，主从分离与否没有绝对的优劣，它仅仅是一种架构设计，各自有适用的场景；读写分离应用场景：读多写少。

10、 Kafka-实战中踩坑好文推荐

案例:

转载地址：http://euacz.baihongyu.com/