kafka

kafka由多个broker组成，每个broker是一个节点（伪集群模式下，多个broker运行在一个机器上时，一个broker就是一个进程）

borker负责接收和处理客户端发送过来的请求，以及对消息进行持久化

承载消息的逻辑容器，在实际业务中多用来区分具体的业务

 一个topic可以划分为多个partition，每个partition可以存在于不同的broker上

每个分区是一组有序的消息日志

生产者生产的每条消息只会发送到一个分区中

副本概念

副本工作机制

每个消息被添加到分区时，分配唯一的offset，以此保证partition内消息的顺序性

分区位移总是从0开始，假设一个生产者向一个空分区写入了10条消息，那么这10条消息的位移依次是0、1、2、...、9

多个消费者实例共同组成的一个组，同时消费多个分区以实现高吞吐

每个Consumer属于一个特定的Consumer Group，一条消息可以发送到多个不同的Consumer Group，但是一个Consumer Group中只能有一个Consumer能够消费该消息

表示消费者消费进度，每个消费者都有自己的消费者位移

消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程

Rebalance是kafka消费者端实现高可用的重要手段

向Topic发送新消息的应用程序

从Topic订阅新消息的应用程序

AR（Assigned Replicas）

ISR（In-Sync Replicas）

OSR（Out-of-sync replicas）

AR = ISR + OSR

log.dir

log.dirs

zookeeper.connect

listeners（内网）

advertised.listeners（外网）

auto.create.topics.enable

unclean.leader.election.enable

auto.leader.rebalance.enable

retention.ms

retention.bytes

log.retention.{hour|minutes|ms}

log.retention.bytes

message.max.bytes

堆大小

垃圾回收器（GC）

如何设置

文件描述符限制

文件系统类型

Swappiness

提交时间（Flush落盘时间）

0.8版本之前，是无HA机制的

0.8版本之后，提供了HA机制，增加了replica(复制品)副本机制

副本定义

副本机制原理

副本机制的好处

等价于日志项（record item），在里层

等价于消息批次（Record batch）, 在外层，里面包含若干条消息

Producer以record batch为单位发送消息

时间换空间，用CPU时间去换磁盘空间或网络IO传输量

kafka在消息头部添加了一个描述压缩属性字节，这个字节的后两位表示消息压缩采用的编码，如果后两位是0，表示消息未被压缩

生产者端

Broker端

Broker端

消费者端

压缩算法的存储

分类

评价指标

性能比较

开启压缩

解压缩

在Kafka文件存储中，同一个topic下有多个不同partition，每个partition为一个目录，partiton命名规则为 topic名称+有序序号

第一个partiton序号从0开始，序号最大值为partitions数量减1

每个partion(目录)相当于一个巨型文件被平均分配到多个大小相等segment(段)数据文件中

每个partiton只需要支持顺序读写就行了，segment文件生命周期由服务端配置参数决定

这样做的好处就是能快速删除无用文件，有效提高磁盘利用率

public interface ProducerInterceptor<K, V> extends Configurable {
/**
* 消息发送前调用
*/
public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
/**
* 消息发送后，服务器返回结果（成功或错误）时调用
*/
public void onAcknowledgement(RecordMetadata metadata, Exception exception);

/**
* 拦截器关闭时调用
*/
public void close();
}

public class ProducerInterceptorPrefix implements ProducerInterceptor<String,String> {
private volatile long sendSuccess = 0;
private volatile long sendFailure = 0;
@Override
public ProducerRecord<String, String> onSend(ProducerRecord&<String, String> record) {
String modifiedValue = "prefix1-" + record.value();
return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(),
record.key(), modifiedValue,
record.headers());
}
@Override
public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
if (e == null) {
sendSuccess++;
} else {
sendFailure ++;
}
@Override
public void close() {
double successRatio = (double)sendSuccess / (sendFailure + sendSuccess);
System.out.println("[INFO] 发送成功率=" + String.format("%f", successRatio * 100) + "%");   
}
}

properties.put(ProducerConfig.INTERCEPTOR_CLASS_CONFIG，ProducerInterceptorPrefix.class.getName())

一个topic可以划分为多个partition

每条消息只会保存在某一个分区中，而不会在多个分区中被保存多份

为消息分配分区

实现消息的顺序读写

提供负载均衡的能力（实现系统的高伸缩性）

消息在经过序列化之后，就需要确定它发往的分区，如果ProducerRecord中指定了partition字段（指定了推送的分区），就不需要分区器的作用，因为partition就是要发往的分区编号

概念

算法

当一个主题被创建后，kafka只支持分区的增加，不支持减少

不支持减少分区的原因

替代方案

分区是实现负载均衡以及高吞吐量的关键，故在生产者这一端就要仔细盘算合适的分区策略，避免造成消息数据的倾斜，使得某些分区称为性能瓶颈，这样极易引发下游数据消费的性能下降

kafka所有的通信都是基于TCP连接的

1. KafkaProducer 实例创建时启动 Sender 线程，从而创建与 bootstrap.servers 中所有 Broker 的 TCP 连接

2. KafkaProducer 实例首次更新元数据信息之后，还会再次创建与集群中所有 Broker 的 TCP 连接

3. 如果 Producer 端发送消息到某台 Broker 时发现没有与该 Broker 的 TCP 连接，那么也会立即创建连接

用户主动关闭

Kafka自动关闭

定义：
Reactor模式是事件驱动架构的一种实现方式，特别适合应用于多个客户端并发向服务器端发送请求的场景

架构图

架构图

请求处理架构图

Kafka的消费者采用从broker 拉取(pull) 消息的方式获取消息

Kafka消费消息是一个不断轮询的过程，消费者只要不断重复调用poll方法即可， poll() 方法返回由生产者写入 Kafka 但还没有被消费者读取过的记录

KafkaConsumer是非线程安全的（但线程架构），poll方法中定义了一个acquire方法，用来检测当前是否只有一个线程在操作；
如果有其他线程操作会抛出ConcurrentModificationException异常
acquire方法中并没有调用锁（synchronized或juc的lock）, 它通过线程操作技术标记的方式来检测线程是否发生了并发操作，以此保证只有一个线程在操作

Kafka集群保持所有的消息，直到它们过期（无论消息是否被消费）。实际上消费者所持有的仅有的元数据就是这个offset（偏移量），也就是说offset由消费者来控制：正常情况当消费者消费消息的时候，偏移量也线性的的增加。但是实际偏移量由消费者控制，消费者可以将偏移量重置为更早的位置，重新读取消息。可以看到这种设计对消费者来说操作自如，一个消费者的操作不会影响其它消费者对此log的处理

概念

原理

这个模型可能存在多个发布者向相同的主题(Topic)发送消息，而订阅者也可以存在多个，他们都能接收到相同主题的消息

public interface ConsumerInterceptor<K, V> extends Configurable {
     public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records);
     public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets);
     public void close();
}

public class ConsumerInterceptorTTL implements ConsumerInterceptor<String, String> {
private static final long EXPIRE_INTERVAL = 10000;

@Override
public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
long now = System.currentTimeMillis();
Map<TopicPartition, List> ConsumerRecord<String, String >>> newRecords = new HashMap<>();
for (TopicPartition tp : records.partitions()) {
List<org.apache.kafka.clients.consumer.ConsumerRecord<String, String>> newTpRecords = new ArrayList<>();
List<ConsumerRecord<String, String>> tpRecords = records.records(tp);
for (ConsumerRecord<String, String> record : tpRecords) {
if (now - record.timestamp() < EXPIRE_INTERVAL) {
newTpRecords.add(record);
}
}
newRecords.put(tp, newTpRecords);
}
return new ConsumerRecords<>(newRecords);
}

@Override
public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
offsets.forEach((tp, offset) -> System.out.println(tp + ":" + offset.offset()));
}

@Override
public void close() {
}

@Override
public void configure(Map<String, ?> configs) {
}
}

消息组（Consumer Group）是一个逻辑上的概念，它将消费者归为一类；每一个消费者只属于一个消费组；
Consumer Group 是kafka提供的可扩展且具有容错性的消费者机制

每个消费组都会有一个固定的名称，消费者在进行消费前需要指定所属消费者的名称；
参数 group.id，默认为空字符串

每个消费者都有一个对应的消费组，当消费发布到主题后，只会被投递给订阅它的每个消费组中的一个消费者

如果所有实例都属于同一个Group, 那么它实现的就是消息队列模型；如果所有实例分别属于不同的Group，那么它实现的就是发布/订阅模型

机制

Range strategy（范围分区）

RoundRobin strategy（轮询分区）

消费者组的模型可以让整体的消费能力具备横向伸缩性

分区分配的演变

通过增加（减少）消费者的个数来提高（降低）整体的消费能力

比如某个Group下有20个Consumer实例，它订阅了1个具有100个分区的Topic。
正常情况下，kafka平均会为每个Consumer分配5个分区。
这个分配的过程就叫Reblance

1. （被动/故障导致）组成员数发生变更：有新的Consumer实例加入组或者离开组，或者有Consumer实例崩溃被"踢出"组

2. （主动运维操作）订阅主题数发生变更: 当Consumer Group使用正则表达式的方式订阅主题，在Consumer Group运行过程中，新创建了一个满足这样条件的主题，则该Group就会发生Rebalance

3.  (主动运维操作) 订阅主题的分区数发生变更：当增加主题的分区数时，会触发订阅这个主题的Group开启Rebalance

Consumer Group 下有两个 Consumer，比如 A 和 B，当第三个成员 C 加入时，Kafka 会触发 Rebalance，并根据默认的分配策略重新为 A、B 和 C 分配分区

概念：
在 Rebalance 过程中，所有 Consumer 实例共同参与，在协调者组件的帮助下，完成订阅主题分区的分配

影响点：
在整个Rebalance过程中，所有实例都不能消费任何消息，因此对Consumer的TPS影响很大

协调者（Coordinator）

影响Consumer端TPS:
在Rebalance中，所有Consumer实例都会停止消费，等待Rebalance

Rebalance过程很慢:
如果Group下Consumer实例过多，Rebalance过程很慢（可能长达几小时）

Rebalance效率不高：
目前Rebalance的设计是所有Consumer实例共同参与，全部重新分配所有分区；
（没法复用以前负责的分区，不会保留以前的分配方案，而是打乱全部重新分配）

重复消费：
消费者消费完某个分区的一部分消息还没有来得及提交，发生了Rebalance;
这个分区被分配给Group中的另一个消费者，原来被消费的消息会被重新消费一遍

   第一种情况：
（订阅主题数量发生变化，订阅主题的分区数发生变化），这2个都是主动运维操作引发的Rebalance，都是不可避免的，也是可以接受的

第二种情况:
Consumer Group 下的 Consumer 实例数量发生变化（最常见的Rebalance）

第一种情况：
未能及时发送心跳，导致Consumer被踢出Group，而引发Rebalance

第二种情况:
Consumer 消费时间过长

第三种情况：
Consumer端的频繁GC

原理:
靠的是消费者的心跳线程

机制：
当协调者决定开启新一轮重平衡后，它会将“REBALANCE_IN_PROGRESS”封装进心跳请求的响应中，发还给消费者实例。当消费者实例发现心跳响应中包含了“REBALANCE_IN_PROGRESS”，就能立马知道重平衡又开始了，这就是重平衡的通知机制

定义：
kafka设计了一套消费组状态机，来帮助协调者完成整个重平衡流程

Kafka 为消费者组定义了 5 种状态

消费者端Rebalance流程

Borker端（协调者端）Rebalance流程

以前版本

1.0.9版本以后

消息格式（三种）

创建时机

位移提交

位移过期消息的删除

找不到消费位移

和生产者不同，创建KafkaConsumer实例时是不会创建任何TCP连接的

TCP连接是在调用KafkaConsumer.poll 方法时被创建的；
在poll方法内部有3个时机可以创建TCP连接

主动关闭

自动关闭

定义：
对Producer和Consumer要处理的消息提供什么样的承诺

常见的承诺

kafka默认提供的交付可靠性保证是至少一次

当Prodicer无法确定消息是否成功提交时，会选择重试（可能导致重复发送）（关闭Producer重试即可实现最多一次）

幂等性Producer

事务型Producer

作用（如何选取）

不足

发后既忘（fire and forget）

同步（sync）

异步（async）

可重试的异常

不可重试的异常

配置

原理

配置

配置

Kafka 只对“已提交”的消息（committed message）做有限度的持久化保证

生产者端：使用带回调的API / acks=all / retries=MAX

kafka服务器端：unclean.leader.election.enable=false/ replication.factor >1 / min.insync.replicas >1

消费者端：enable.auto.commit=false（手动提交offset）

机制

消息丢失的场景

如何保证消息不丢失

设置enable.auto.commit为false

关闭自动提交位移

一方面需要消息中间件来进行保证。另一方面需要自己的处理逻辑来保证消息的幂等性

有可能代码消费了消息，但服务器突然宕机，未来得及提交offset。所以我们可以在代码保证消息消费的幂等性

可以通过redis的原子性来保证，也可以通过数据库的唯一id来保证

考量点

建议

考量点

建议

考量点

建议

考量点

建议

控制器帮助完成对Kafka主题的创建、删除以及增加分区的操作；

执行Kafka-topics.sh脚本时大部分工作都是由控制器完成

主要指使用 kafka-reassign-partitions 脚本 对已有分区进行细粒度的分配功能

kafka为了避免部分Broker负载过重而提高的一种换Leader的方案

自动检测新增Broker、Broker主动关闭及Broker宕机等（依赖Zookeeper的Watch功能和临时节点组合实现）

示例

向其他Broker提高数据服务

控制器上保存了最全的集群元数据，控制器会定期地往其他Broker发送请求，从而更新自己内存中的元数据

所有主题信息: 
包括具体的分区信息，比如领导者副本是谁，ISR集合中有哪些副本等

所有Broker信息：
包括当前都有哪些运行中的Broker，哪些正在关闭中的 Broker等

所有涉及运维任务的分区:
包括当前正在进行 Preferred 领导者选举以及分区重分配的分区列表

zookeeper通过Watch机制快速感知到控制器所在的Broker宕机，并马上删除 /controller 节点

存活的Broker开始竞争新的控制器，第一个在zookeeper上创建 /controller 节点的Broker当选控制器

新的控制器从zookeeper中读取集群元数据信息，并初始化到自己的内存中；Failover 完成

1. Kafka使用消息数据（Log）来保存数据，一个日志就是磁盘上一个只能追加写（Append-only）消息的物理文件

2. 因为只能追加写入，故避免了缓慢的随机I/O操作，改为性能较好的顺序I/O写操作，这也是kafka实现高吞吐量的一个手段

原理就是通过日志段（Log Segment）机制

在kafka底层，一个日志又近一步细分成多个日志段，消息被追加写到当前最新的日志段中，当写满一个日志段后，kafka会自动切分出一个新的日志段，并将老的日志段封存起来。kafka在后台还有定时任务会定期地检查老的日志段是否能够被删除，从而实现回收磁盘空间的目的

它是一个分布式协调框架，负责协调管理并保存 Kafka 集群的所有元数据信息，比如集群都有哪些 Broker 在运行、创建了哪些 Topic，每个 Topic 都有多少分区以及这些分区的 Leader 副本都在哪些机器上等信息

HW

LEO

每个副本对象都保存了一组高水位值和LEO值；
在Leader副本所在的Broker上，还保存了其他Follower副本的LEO值

follower副本何时更新LEO

follower副本何时更新HW

leader副本何时更新LEO

leader副本何时更新HW值

当尝试确定分区HW时，它会选出所有满足条件的副本，比较它们的LEO（当然也包括leader自己的LEO），并选择最小的LEO值作为HW值

满足条件

初始状态

场景一：follower发送FETCH请求，在leader处理完PRODUCE请求之后

场景二：FETCH请求保存在purgatory中PRODUCE请求到来

备份数据丢失

备份数据不一致

方案

数据结构
Leader Epoch由两部分数据组成

举例

原理

举例说明（看上图）

broker

分区

topic

每个Broker在启动时，都会到Zookeeper上进行注册，即到/brokers/ids下创建属于自己的节点，如/brokers/ids/[0...N]

zookeeper会通过Watch机制将消息推送给控制器

每个Broker就会将自己的IP地址和端口信息记录到该节点中去

其中，Broker创建的节点类型是临时节点，一旦Broker宕机，则对应的临时节点也会被自动删除

当Broker在启动时，都会尝试去zookeeper上创建 /controller 节点；

第一个成功创建 /controller 节点的Borker就是当前集群的控制器

故障转移：

在Kafka中，同一个Topic的消息会被分成多个分区并将其分布在多个Broker上，这些分区信息及与Broker的对应关系也都是由Zookeeper在维护，由专门的节点来记录，如：/borkers/topics

生产者需要将消息合理地发送到这些分布式的Broker上；

Kafka支持传统的四层负载均衡，也支持Zookeeper方式实现负载均衡

每条消息都只会发送给分组中的一个消费者

不同的Consumer Group消费自己特定的Topic下面的消息，互不干扰

在Kafka中，规定了每个消息分区 只能被同组的一个消费者进行消费

因此，需要在 Zookeeper 上记录 消息分区 与 Consumer 之间的关系，每个消费者一旦确定了对一个消息分区的消费权力，需要将其Consumer ID 写入到 Zookeeper 对应消息分区的临时节点上，例如：
/consumers/[group_id]/owners/[topic]/[broker_id-partition_id]

在消费者对指定消息分区进行消息消费的过程中，需要定时地将分区消息的消费进度Offset记录到Zookeeper上，以便在该消费者进行重启或者其他消费者重新接管该消息分区的消息消费后，能够从之前的进度开始继续进行消息消费

/consumers/[group_id]/offsets/[topic]/[broker_id-partition_id]

1.0.9版本以后，修改为保存到kafka内部（将Consumer的位移数据作为一条条普通的kafka消息，提交到_consumer_offset中）

每个消费者服务器启动时，都会到Zookeeper的指定节点下创建一个属于自己的消费者节点。例如/consumers/[group_id]/ids/[consumer_id]

完成节点创建后，消费者就会将自己订阅的Topic信息写入该临时节点