Kafka 权威指南阅读笔记(第三章,第四章)

时间:2022-06-02 18:08:48

Kafka 第三章,第四章阅读笔记

  1. Kafka 发送消息有三种方式:不关心结果的,同步方式,异步方式。
  2. Kafka 的异常主要有两类:一种是可重试异常,一种是无需重试异常。

生产者的配置:

  1. acks 越大,越不容易丢失消息,但是吞吐量下降。
  2. buffer.memory,设置不当会导致阻塞或者抛出异常。
  3. compression.type snappy和gzip, lz4。
  4. retries 重试次数,
  5. 如果要保证消息的顺序,必须保证max.in.flight.requests.per.connection=1.

分区

1.一个进程只从一个分区读取数据
2.相同键的消息都会被写到同一个分区。
3.使用默认分区器时,只有不改变分区数据的情况下,键和分区的映射关系才能保持不变。
4.群组协调器broker,不同的群组可以有不同的协调器。消费者在轮询消息或者提交偏移量时会发送心跳,如果心跳长时间停止,会话会过期,群组协调器会认为消费者死亡,发生再均衡。
5.同一个Topic的分区,只被同一个消费组里的消费者拥有一次。有多个放映厅(消费组),一场电影(分区),每一个放映厅里面人只能看一次。一个放映厅的人可以看多个不同的电影。

轮询间隔:
max.poll.interval.ms

分配分区:

消费者第一个加入群组,则它将成为群主。
群主从协调器获得群组的成员列表,并负责给每一个消费者分配分区。实现了一个PartitionAssignor的接口。
Kafka实现了两种分配策略。
分配完毕之后,群主把分配情况发送给群组协调器,协调器再把信息发送给所有的消费者。只有群主知道群组里所有的消费者信息。
该过程在每次再均衡时发生。

消费者的配置:

fetch.min.bytes,如果数据量小于fetch.min.bytes指定的大小,会等待满足条件broker才会返回给消费者。
fetch.max.wait.ms,指定broker的等待时间,默认500ms,如果没有足够的数据流入Kafka,会导致500ms的延迟。
如果fetch.max.wait.ms=100ms,fetch.min.bytes=1MB,Kafka收到消费者的请求后,要么返回1MB的数据,要么100ms后返回所有的数据。
max.partition.fetch.bytes 默认值是1MB。 poll()方法从每个分区里返回的记录最多不超过这个值。20分区,5个消费者,则至少4MB可用内存才能接收记录。该值必须比broker能手接受的最大消息的字节数,max.message.size大,否则可能无法读取消息,导致消费者一直挂起重试。。。
如果单次返回的数据过多,可能来不及处理数据,导致会话过期,这种情况下,可以把该值改小,或者延长会话过期时间。

session.timeout.ms
消费者超过这个时间没有发送心跳给群组协调器则会被认为死亡,从而触发再均衡。
该属性和heartbeat.interval.ms紧密相关,heartbeat.interval.ms指定了poll()方法向协调器发送心跳的频率。session.timeout.ms则指定了消费者多久可以不发心跳。
一般来说,heartbeat.interval.ms必须比session.timeout.ms小,一般是1/3。session.timeout.ms属性太小,有利于更快检测和恢复崩溃的节点,但是长时间轮询或垃圾收集等情况会导致非预期的再均衡。相反该值过大,可以减少意外的再均衡,但是检测崩溃节点会需要更长的时间。

auto.offset.reset
在偏移量无效的情况下,消费者将从何处读取分区记录。因消费者长时间失效,包含偏移量的记录已经过时并被删除。默认值为latest,意思为从最新的记录开始读取。另一个值是earliest,从起始位置读取分区的记录。

enable.auto.commit
该属性指定消费者是否自动提交偏移量,默认是true。为了尽量避免重复数据和数据丢失,可以把他设置为false,由自己控制何时提交偏移量。
auto.commit.interval.ms 可以控制相应的自动提交的频率,默认值是5秒。

partition.assignment.strategy
分区分配策略,kafka内置两种策略:
Range:该策略会把连续的分区分配给消费者。当分区数量是奇数时,容易将更多的分区分配给某一个消费者。
RoundRobin:该策略会将分区逐个分配给消费者,如果所有的消费者都订阅相同的主题,所有消费者分配到数量相同的分区,最多相差一个。

client.id
客户端的ID

max.poll.records
用于控制单次调用call方法能够返回的记录数量。

receive.buffer.bytes和send.buffer.bytes
socket用到的TCP的缓冲区大小。-1表示使用操作系统的默认值。

消费者如何提交偏移量

消费者更新分区的当前位置的操作叫做提交。
消费者往一个叫做_consumer_offset的特殊主题提交偏移量。
自动提交,按照时间间隔。
手动提交:
同步提交,异步提交。最好的写法就是轮询中异步提交,再最外部,关闭之前同步提交。

再均衡监听器

可以在发生再均衡期间提交偏移量。实现 ConsumerRebalanceListener接口。

  1. onPartitionsRevoked 方法会在再均衡开始之前,且消费者停止读取消息之后被调用。
  2. onPartitionAssigned 方法会在重新分配分区之后,且消费者开始毒气消息之前被调用。

seek 可以自定义偏移量。

退出

主线程上的消费者,退出线程可以使用addShutdownHook。退出循环的安全方式是通过wakeup()方法。poll会抛出WakeupException,退出循环。在退出之前,确保关闭了消费者。

没有群组的消费者。独立消费者

一个消费者可以订阅主题加入消费组,或者为自己分配分区,但是不能同时做这两件事情。

步骤是,先从集群请求主题哪些分区可用,然后调用assign方法。