首页  思维导图  详情

kafka

2020-06-23 18:47:54   81  举报





AI智能生成

kafka简介

kafka

作者其他创作

大纲/内容

设计特点

高吞吐量，每秒百万级的消息读写

持久化存储，并有replication副本防止丢失

分布式，producer，broker，consumer都为分布式实现

磁盘系统

kafka的高吞吐量主要得益于磁盘系统的使用设计，broker基于内存映射文件
技术，将消息先写入系统的页缓存中，页缓存直接映射到磁盘文件

架构

图

Kafka强依赖于ZooKeeper

brocker

broker就是kafka server，每一台kafka服务器都是一个brocker

选举

通过在zk上创建临时节点/controller临时节点来实现leader选举，并在该节点上写入当前brocker信息，
利用zk的强一致性特性，一个节点只能被一个客户端创建成功，创建成功的broker即为leader，即先到先得原则，
其他节点会监听watch /controller节点

producer

主要负责序列化消息并发送

consumer

consumerGroup

消费者组，含有一个组名，partition在一个组内只能有一个consumer消费，
可以用来实现消息广播，多个CG同时订阅一个topic

子主题

topic

主题，是写入kafka的基础单元，是一个逻辑概念,实际写入的是partition，一个topic包含多个partition

每条消息属于且仅属于一个Topic

发送和订阅消息都必须指定topic

partition

topic的组成单元，可以水平扩容，是kafka高吞吐量的保障

消息持久化时，每条消息都是根据一定的分区规则路由到对应的partition中，并append到log文件的尾部

在同一个partition中消息是顺序写入且有序的，但不同partiton之间不能保证消息的有序性

个数最好与服务器个数相当

多个大小相等的segment file (段)组成了一个partition

副本 replication

partition可以有指定数据的副本，主从模式，producer和consumer只与leader交互
follower从leader复制

ISR

in-sync replica，已同步的副本

Kafka会在Zookeeper上针对每个Topic维护一个ISR，持有partition的已同步的副本信息，
如果某个分区的Leader不可用，Kafka就会从ISR集合中选择一个副本作为新的Leader。

setment file

每个partition 就相当于一个巨型的文件里面由多个大小相等的segment file小文件组成，
但是每个segment file 的消息数量并不一定相等，

组成

.index 索引文件

包含若干索引条目，每个条目表示数据文件中一条message的索引

.log 数据文件

offset

位移

partition中的每个消息都有一个连续的序号，用于partition唯一标识一条消息。
Offset记录着下一条将要发送给Consumer的消息的序号。
Offset从语义上来看拥有两种：Current Offset和Committed Offset。

current offset

Current Offset保存在Consumer中，它表示Consumer希望收到的下一条消息的序号。
它仅仅在poll()方法中使用。例如，Consumer第一次调用poll()方法后收到了20条消息，那么Current Offset就被设置为20。
这样Consumer下一次调用poll()方法时，Kafka就知道应该从序号为21的消息开始读取。这样就能够保证每次Consumer poll消息时，
都能够收到不重复的消息。

Committed Offset

已提交位移，保存在Broker上，表示Consumer已经确认消费过的消息的序号，举个例子，Consumer通过poll() 方法收到20条消息后，此时Current Offset就是20，经过一系列的逻辑处理后，并没有调用consumer.commitAsync()或consumer.commitSync()来提交Committed Offset，那么此时Committed Offset依旧是0。

总结

Current Offset是针对Consumer的poll过程的，它可以保证每次poll都返回不重复的消息；
而Committed Offset能够保证新的Consumer能够从正确的位置开始消费一个partition，从而避免重复消费。

存储模型

groupid-topic-partition -> offset的方式保存。实际上保存在__consumers_offsets这个topic中。

参数配置

ACKS

影响消息持久化

auto.offset.reset

表示如果Kafka中没有存储对应的offset信息的话（有可能offset信息被删除），消费者从何处开始消费消息。它拥有三个可选值：

earliest：从最早的offset开始消费
latest：从最后的offset开始消费
none：直接抛出exception给consumer

场景

Consumer消费了5条消息后宕机了，重启之后它读取到对应的partition的Committed Offset为5，
因此会直接从第6条消息开始读取。此时完全依赖于Committed Offset机制，和auto.offset.reset配置完全无关。

新建了一个新的Group，并添加了一个Consumer，它订阅了一个已经存在的Topic。
此时Kafka中还没有这个Consumer相应的Offset信息，因此此时Kafka就会根据auto.offset.reset
配置来决定这个Consumer从何处开始消费消息。

producer配置

bootstrap.servers	主机，配置格式： host1:port1,host2:port2,.... 由于这些主机是用于初始化连接，以获得整个集群（集群是会动态变化的），因此这个配置清单不需要包含整个集群的服务器。（当然，为了避免单节点风险，这个清单最好配置多台主机）。			high
key.serializer	关键字的序列化类，实现以下接口： org.apache.kafka.common.serialization.Serializer 接口。			high
value.serializer	值的序列化类，实现以下接口： org.apache.kafka.common.serialization.Serializer 接口。			high
acks	消息持久化级别配置，见详细介绍	1	[all, -1, 0, 1]	high
buffer.memory	Producer 用来缓冲等待被发送到服务器的记录的缓存。如果记录发送的速度比发送到服务器的速度快， Producer 就会阻塞，如果阻塞的时间超过 max.block.ms 配置的时长，则会抛出一个异常。	32MB	[0,...]	high
retries	若设置大于0的值，则客户端会将发送失败的记录重新发送，尽管这些记录有可能是暂时性的错误。请注意，这种 retry 与客户端收到错误信息之后重新发送记录并无区别。允许 retries 并且没有设置max.in.flight.requests.per.connection 为1时，消息可能乱序。比如：当两个批次都被发送到同一个 partition ，第一个批次发生错误并发生 retries 而第二个批次已经成功，则第二个批次的记录就会先于第一个批次出现。	0	[0,...,2147483647]	high
batch.size	producer会把发往同一分区的多条消息封装进一个batch中，当batch满了后，producer才会把消息发送出去小的 batch.size 降低吞吐量(如果 batch.size = 0的话将完全禁用批处理)。很大的 batch.size 可能造成内存浪费	16KB	[0,...]	medium
linger.ms	producer 会将两个请求发送时间间隔内到达的记录合并到一个单独的批处理请求中。通常只有当记录到达的速度超过了发送的速度时才会出现这种情况。producer 将等待给定的延迟时间，以便将在等待过程中到达的其他记录能合并到本批次的处理中。但batch.size达到设置值时，Producer 会忽略这个参数，立刻发送数据。但是如果累积的字节数少于 batch.size ，那么将在指定的时间内“逗留”(linger)，以等待更多的记录出现。这个设置默认为0(即没有延迟)。例如：如果设置linger.ms=5 ，则发送的请求会减少并降低部分负载，但同时会增加5毫秒的延迟。	0	[0,...]	medium
max.request.size	请求的最大字节数。这个设置将限制 Producer 在单个请求中发送的记录批量的数量，以避免发送巨大的请求。这实际上也等同于批次的最大记录数的限制。请注意，服务器对批次的大小有自己的限制，这可能与此不同。	1MB	[0,...]	medium

consumer配置

bootstrap.servers	主机，配置格式： host1:port1,host2:port2,.... 由于这些主机是用于初始化连接，以获得整个集群（集群是会动态变化的），因此这个配置清单不需要包含整个集群的服务器。（当然，为了避免单节点风险，这个清单最好配置多台主机）。			high
key.serializer	关键字的序列化类，实现以下接口： org.apache.kafka.common.serialization.Serializer 接口。			high
value.serializer	值的序列化类，实现以下接口： org.apache.kafka.common.serialization.Serializer 接口。			high
fetch.min.bytes	获取请求返回的最小数据量。如果没有足够的数据可用，请求将等待这么多数据累积后才响应请求。默认设置为1个字节意味着一旦有一个字节的数据可用，或者fetch请求等待数据到达的时间过长，fetch请求就会得到响应。将此值设置为大于1将导致服务器等待更大数量的数据累积，这可以稍微提高服务器的吞吐量，但代价是增加一些延迟。	1	[0,...]	high
group.id	标识所属的消费者组的ID。如果consumer通过使用subscribe(topic)或基于kafka的偏移管理策略来使用组管理功能，则需要此属性。	""		high
heartbeat.interval.ms	心跳用于确保消费者会话保持活动状态，并在新消费者加入或离开组时重新平衡。该值必须设置为低于session.timeout。，但一般应设置不高于该值的1/3。它可以调整甚至更低，以控制正常再平衡的预期时间。	3000		high
max.partition.fetch.bytes	限制Consumer每次发起fetch请求时，读取到的数据大小	1MB	[0,...]	high
session.timeout.ms	Consumer session 过期时间。这个值必须设置在broker configuration中的group.min.session.timeout.ms 与 group.max.session.timeout.ms之间。	10s		high

enable.auto.commit	自动提交offset	true		medium
auto.commit.interval.ms	offset自动提交到Kafka的频率。	5s		low