kafka学习分享（浅尝版）

初识kafka参考：1

目录：

1、消费和生产一些常用参数和情况

2、rebalance机制

3、高性能的原因

4、消息丢失场景及方案

5、注意（遇到的报错等）

一、KafkaConsumer KafkaProducer demo Properties props = new Properties(); //逗号分隔，可以不指定全 props.put("bootstrap.servers", "broker1:9092, broker2:9092"); //默认值是""，显示指定不然InvalidGroupldException props.put("group.id", "groupId666"); //无默认值，必须指定，不然ConfigException props.put("key.deserializer", "org.apache.kafka.common.serializaiton.StrignDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serializaiton.StrignDeserializer"); //earliest：指定从最早的位移开始消费 。 注意这里最早的位移不一定就是 0 。 //latest：指定从最新处位移开始消费 。 //none ：指定如果未发现位移信息或位移越界，则抛出异常。 props.put("auto.offset. reset", "earliest"); //默认500 props.put("max.poll.records", "100"); //自动提交默认5s,也可自己设置auto.commit.interval.ms，设置大点防止rebalance //false的话就自己提交consumer.commitSync(); consumer.commitAsync(); props.put("enable.auto.commit", "true"); //session.timeout.ms : 默认3s //heartbeat.interval.ms :1s //fetch.min.bytes : 降低broker工作负载 //max.poll.interval.ms 单次poll()操作后可以执行的最长时间，或者poll()调用之间的最大延迟，如果在这个延迟时间之内未收到下一次poll()操作，将认为客户端已失败，从而触发Rebalance操作，默认时间为5分钟，tips: 该值设置不正确时可能无法消费延迟消息; KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); consumer.subscribe(Collections.singletonList("customerCountries")); ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1L)); for (ConsumerRecord<String, String> record: records) { // ... } Properties props = new Properties(); props.put("bootstrap.servers", "broker1:9092, broker2:9092"); props.put("acks", "all");//所有follower都响应了才认为消息提交成功，即"committed" props.put("retries", 0);//max/-1、1、0 props.put("batch.size", 16384);//producer将试图批处理消息记录，以减少请求次数.默认的批量处理消息字节数16KB //batch.size当批量的数据大小达到设定值后，就会立即发送，不顾下面的linger.ms props.put("linger.ms", 1);//延迟1ms发送，这项设置将通过增加小的延迟来完成--即，不是立即发送一条记录，producer将会等待给定的延迟时间以允许其他消息记录发送，这些消息记录可以批量处理 props.put("buffer.memory", 33554432);//producer可以用来缓存数据的内存大小。一个线程发、一个线程在内存缓存 props.put("key.serializer","org.apache.kafka.common.serialization.IntegerSerializer"); props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer"); KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props); producer.send(new ProducerRecord<>(topic, data))

（名词扫盲）

Broker：Kafka集群包含一个或多个服务器，这种服务器被称为broker Producer：负责发布消息到Kafka broker Topic：每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic（类似MQ中的queue） Partition：每个Topic包含一个或多个Partition Segment：多个大小相等的段组成了一个partition，segment文件名为上一个segment文件最后一条消息的offset值。一个segment里面有2个文件，一个是.log文件，它是topic数据存储的文件，还有一个文件叫.index文件，它是.log文件索引文件。 Offset：一个连续的用于定位被追加到分区的每一个消息的序列号，最大值为64位的long大小，19位数字字符长度。 Consumer：消息消费者，向Kafka broker读取消息的客户端。 Consumer Group：每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group） Group Coordinator是一个服务，每个Broker在启动的时候都会启动一个该服务。Group Coordinator的作用是用来存储Group的相关Meta信息，并将对应Partition的Offset信息记录到Kafka内置Topic(__consumer_offsets)中。Kafka在0.9之前是基于Zookeeper来存储Partition的Offset信息(consumers/{group}/offsets/{topic}/{partition})，因为ZK并不适用于频繁的写操作，所以在0.9之后通过内置Topic的方式来记录对应Partition的Offset。 Kafka，ActiveMQ，RabbitMQ，RocketMQ 对比

ActiveMQ：JMS规范，支持事务、支持XA协议，没有生产大规模支撑场景、官方维护越来越少

RabbitMQ：erlang语言开发、性能好、高并发，支持多种语言，社区、文档方面有优势，erlang语言不利于java程序员二次开发，依赖开源社区的维护和升级，需要学习AMQP协议、学习成本相对较高以上吞吐量单机都在万级

kafka：高性能，高可用，生产环境有大规模使用场景，单机容量有限（超过64个分区响应明显变长）、社区更新 吞吐量单机百万

rocketmq：java实现，方便二次开发、设计参考了kafka，高可用、高可靠，社区活跃度一般、支持语言较少吞吐量单机十万

二、rebalance机制

时机：增加或减少group中consumer的时候会触发rebalance；consumer消费超时；订阅topic变化；订阅topic的分区变化。

一、coordinator：通常是partition的leader节点所在的broker，负责监控group中consumer的存活，consumer维持到coordinator的心跳，判断consumer的消费超时 1、coordinator通过心跳返回通知consumer进行rebalance

2、consumer请求coordinator加入组，coordinator选举产生leader consumer

3、leader consumer从coordinator获取所有consumer，发送syncGroup（分配信息）给到coordinator

4、coordinator通过心跳机制将syncGroup下发给consumer

5、完成rebalance

二、leader consumer监控topic的变化，通知coordinator触发rebalance。

问题：

如果C1消费消息超时，触发rebalance，重新分配后、该消息会被其他消费者消费，此时C1消费完成提交offset. 导致错误

解决：coordinator每次rebalance，会标记一个Generation给到consumer，每次rebalance该Generation会 +1，consumer提交offset，coordinator会比对Generation，不一致则拒绝提交

三、kafka高性能高吞吐的原因

1、磁盘顺序读写：保证了消息的堆积

顺序读写，磁盘会预读，预读即在读取的起始地址连续读取多个页面，主要时间花费在了传输时间，而这个时间两种读写可以认为是一样的。

随机读写，因为数据没有在一起，将预读浪费掉了。需要多次寻道和旋转延迟。而这个时间可能是传输时间的许多倍。

2、零拷贝：避免CPU将数据从一块存储拷贝到另外一块存储的技术

传统的数据复制：

a、读取磁盘文件数据到内核缓冲区

b、将内核缓冲区的数据copy到用户缓冲区

c、将用户缓冲区的数据copy到socket的发送缓冲区

d、将socket发送缓冲区中的数据发送到网卡、进行传输

零拷贝：磁盘文件->内核空间读取缓冲区->网卡接口->消费者进程

3、分区分段+索引

KafkaBmessage消息实际上是分布式存储在一个一个小的segment中的，每次文件操作也是直接操作的segment，为了进一步的查询优化，Kafka又默认为分段后的数据文件建立了索引文件，就是文件系统上的.index文件。这种分区分段+索引的设计，不仅提升了数据读取的效率，同时也提高了数据操作的并行度

4、批量压缩：多条消息一起压缩，降低带宽

5、批量读写

6，直接操作page cache，而不是JVM，避免GC耗时及对象创建耗时，且读写速度更高，进程重启、缓存也不会丢失

四、消息丢失场景及解决 一、生产

1、ack=0，不重试 producer发送消息完，不管结果了，如果发送失败也就去失了。 2、ack=1，leader crash producer发送消息完，只等待lead写入成功就返回了，leader crash了，这时follower没来及同步，消息丢失。 3、unclean.leader.election.enable =true允许选举ISR以外的副本作为leader，会导致数据丢失，默认为false，producer发送异步消息完，只等待lead写入成功就返回了leader crash了，这时ISR中没有follower，leader AOSR中选举，因为OSR中本来落后于Leader造成消息去失。

解决方案： 1、配置：ack=all/-1，tries > 1，unclean.leader.election.enable：false producer发送消息完，等待follower同步完再返回，如果异常则重试，副本的数量可能影响吞吐量。 不允许选举ISR以外的副本作为1eader. 2、配置：min.insync.replicas >1副本指定必须确认写操作成功的最小副本数量。如果不能满足这个最小值，则生产者将引发一个异常《要么是NotEnoughReplicas，要是NotEnoughReplicasAfterAppend）. min.insync.replicas和ack更大的持久性保证，确保如果大多数副本没有收到写操作，则生产者将引发异常。 3、失败的offset单独记录 producer发送消息，会自动重试，遇到不可恢复异常公抛出，这时可以捕获异常记录到数据库或缓存，进行单独处理。

二、消费

先commit再处理消息。如果在处理消息的时候异常了，但是offset 已经提交了，这条消息对于该消费者来说就是丢失了，再也不会消费到了。

三、broker的刷盘

减小刷盘间隔

注意：

1、如果 broker 端没有显式配置 listeners （或 advertised.listeners ）使用 IP 地址的话，那么最好将 bootstrap.servers 配置成主机名而不要使用 IP 地址，因为Kafka 内部使用的是全称域名（ FQDN, Fully Qualified Domain Name ）。倘若不统一 ， 会出现无法获取元数据的异常。

2、LEADER_NOT_AVAILABLE：版本问题；advertised.listeners或listeners配置问题；topic分区是否有对应的brokerId；原来topic在使用中被删除导致的；重启解决一切问题。

3、CommitFailedException:Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group： 报错显示无法正常提交offset了，因为这个消费者已经被踢出消费者组了，服务端以为你挂掉了。 原因1：在session时间内没有收到心跳，以为你挂了。 heartbeat.interval.ms ； session.timeout.ms 设置心跳和session时间，session时间最好是心跳的三倍。 原因2：两次拉取之间时间超过了设置时间，以为你不拉了，也就是消费超时。 可以这样解决： max.poll.interval.ms 设置大一点，多给你点时间处理数据； max.poll.records 设置小一点，消费快点； 原因3：coordinator挂掉了

额，一个批次中一条消息别超过30秒即可（阻塞）。你迟迟不提交offset，可能被其他消费者拿到，造成重复消费。

疑问: 默认心跳时间是3s，session时间是10s，poll间隔时间却是5分钟。显然在我消费的5分钟内，早就过了session时间了，那为啥还能继续下一次poll呢？ 思考： kafka0.10.1之后的版本：session.timeout.ms 和 max.poll.interval.ms 是解耦的，在不断消费过程中，其实有两个线程，一个是心跳线程，一个是消费线程。心跳线程在后台自己维持session内的心跳，那就代表消费者活着；消费线程自己消费完了自己再去poll就好了。然后消费线程，如果过了设置的消费时间，那通过心跳线程认为你活着但是你不消费了，也会把你踢出消费组。 以前的做法（0.10.1版之前）应该是要把session的时间设置的比poll的间隔时间长才行，不然你消费时间很长，超过了session时间，而只有一个线程，没法在你消费的时候发心跳，coordinator协调者就会认为你挂了，直接rebalance。 （不知道理解是否正确，如有不对之处麻烦指正）

---

1.本站遵循行业规范，任何转载的稿件都会明确标注作者和来源；2.本站的原创文章，会注明原创字样,如未注明都非原创，如有侵权请联系删除！；3.作者投稿可能会经我们编辑修改或补充；4.本站不提供任何储存功能只提供收集或者投稿人的网盘链接。