0


kafka使用详解、最佳实践和问题排查

kafka是一个常用的分布式消息中间件,与RabbitMQ对比,特点是可以无限横向扩容,并保持高可靠性、高吞吐量和低延迟,因此比RabbitMQ有更高的市场占有率(网上搜了一下,kafka大约41%,RabbitMQ大约29%)。

一、kafka常见概念

一般正常的开发,了解到前6个概念就好,其余的概念更多用于kafka运维配置,或问题排查。

1、producer生产者

指生产消息,并把消息投递到kafka的外部应用程序 ,它不是kafka的组成部分

2、consumer消费者

指连接到kafka,接收/订阅消息,并进行后续逻辑处理的外部应用程序,它也不是kafka的组成部分。
一个消费者可以同时消费kafka的多个队列(主题)

3、Consumer Group消费者组

连接到kafka的消费者,必须指定消费者组,多个消费者可以指定相同的消费者组,这样可以避免同一个消息被重复消费。
如果2个消费者绑定了同一个队列(主题),指定了不同的消费者组,则每条消息,都会同时投递给这2个消费者。

4、topic 主题

kafka里,收发消息的逻辑集合,每个主题,都可以认为是一个队列;
生产者和消费者,都是通过连接主题 来处理消息。

5、partition分区

kafka里存储消息的物理集合,一个主题可以划分为1个或多个分区,可以理解为子队列;
**每个分区只属于一个主题,且只能被一个消费者消费(同一个组)**。
该主题收到的所有消息,会根据消息的key选择对应的分区进行投递;
如果消息未指定key,且没有定义分区规则时,则kafka会随机平均投递到主题的多个分区里。
注意:每个分区里的消息,一定是按队列的规则,保证先进先出;但是不同的分区的消息,则无法保证。
因此,如果要确保消费者能按消息的投递顺序进行消费:

  • 每个主题只建一个分区(这个不推荐)
  • 同一批需要保证顺序的消息,指定相同的key,比如使用用户ID作为消息的key,相同key的消息会投递到同一个分区

6、offset偏移量

指每个分区里的消息的唯一编号,并且是从0开始递增的。主题+分区+偏移量,可以唯一定位一条消息。
注:每个分区里的每条消息,offset一定是不同的;不同分区的offset是会重复的。
消费者会也记录每次消费的offset值,来标识自己当前处理到哪一条消息了,以便断开重连时,继续消费,消费者的offset也是存储在kafka中。

7、broker集群节点

kafka集群里的某个节点,通常是一台服务器上的kafka实例。

8、replica副本

主题的每个分区,可以指定多个副本,每个副本都存储一份完全相同的消息数据。
一般建议同一个分区的不同副本,要保存在不同的broker上,避免broker故障导致该分区数据丢失。

9、leader/follower

主题的每个分区,如果有多个副本,那么其中一个副本会作为leader对外提供读写服务,其余的作为follower只同步数据。
如果leader出现故障时,会从follower中选举一个副本作为leader重新提供服务。

10、ISR(in-sync replicas)

分区的同步副本集合,每个分区都会维护一个ISR列表,内容是那些与leader保持同步的follower清单。
如果某个follower跟不上同步的进度,或无法保持同步时,会从ISR列表中移除。
只有在ISR列表里的follower,才有机会提升为leader.
注:已经成为leader的副本,也在ISR中

11、LEO日志末端偏移量

LEO指Log End Offset,即当前分区中下一条待写入消息的偏移量,该条消息是未指向具体的消息。
分区的每个副本,都有自己的LEO。

12、HW高水位线

HW指High Watermark Offset,即当前分区中已经被提交并复制到所有副本的最高消息偏移量(offset),
leader接收到消息,但是还未同步完时,不会更新HW值。
leader会比对自己和所有follower的LEO,用其中的较小值,来更新HW值。

13、LAG滞后消息数

一个消费者组,在消费 主题的每个分区时,每个分区都会计算一个LAG值,指该分区的消息总数与消费者已消费的消息数的差值。
通常是 该分区的HW 减 消费者组的offset。
实践中,运维人员应当对LAG进行监控,比如超出10000时进行告警和处理。

理解了这些概念,网上找了一张kafka工作原理图:
在这里插入图片描述

二、kafka与RabbitMQ对比

相对于RabbitMQ,Kafka有如下特点:

  • kafka的消息消费完并不会立即删除,而是保存一定时间后才删除,默认是7天。而RabbitMQ是消费完就删除。 kafka不删消息这点我很喜欢,尤其是排查问题需要数据恢复时。
  • kafka消费者只支持pull模式,不支持push模式,即消费者只能主动轮询kafka获取消息,默认是每500ms拉一次,每次最多拉500条数据,轮询的优点就是灵活,缺点就是没消息时空耗性能。 RabbitMQ默认只支持push模式,主动推送消息给消费者,实时性更好。
  • kafka通过topic主题连接生产者和消费者,多个消费者连接到同一个topic,即可消费该主题的所有消息,如果有不需要的消息,也只能由消费者自行判断和抛弃处理。 RabbitMQ则是通过Exchange接收消息,再通过指定的规则转发到具体的Queue,由消费者消费,可以参考我以前写的文章:https://youbl.blog.csdn.net/article/details/80401945 RabbitMQ可以通过配置很多路由,避免消息投递给不必要的消费者, 不过RabbitMQ也支持直接通过Queue接收和投递消息。
  • 性能上,RabbitMQ是单线程模型,大数据上会有瓶颈;而kafka可以几乎无限扩展。
  • 有序性,kafka对于主题的每个分区,因为有且只能有一个消费者,所以能保证消息的有序性,不同分区则无法保证; 而RabbitMQ在多消费者时,会平均分配消息,无法保证有序,并且在消息消费失败重新投递时,也会破坏消息顺序。

三、kafka最佳实践

1、生产者配置

  • 生产者发送时,有个acks配置,说明如下:- 为0,生产者发消息后,不等borker响应就返回成功,性能最高,丢数据概率也最高;- 为1,生产者发消息后,leader节点返回成功,就算成功;但是leader未同步给其它副本前就挂了,也会丢数据;- 为all 或 -1,则必须等所有副本都同步成功,才返回成功,保证数据不丢失,但性能最低。 生产环境,建议配置为-1,另外2个配置都有丢数据的可能性。
  • min.insync.replicas 最小副本数要求,默认值1,建议为2(当然要求每个topic副本数都在3以上) 因为如果配置为1,当leader收到数据,还未同步就故障了,会丢失数据。
  • retries: 重试次数,设置为较大值,默认值为Integer.MAX_VALUE,确保发送成功。 注:虽然重试次数默认很大,但是重试还受到另一个时间配置的影响:delivery.timeout.ms(默认2分钟),retries还没用完,这个超时时间就到了,也会中断发送。 另外,如果设置了较大的retries,请使用异步发消息的方式,避免同步操作导致线程堵塞,影响用户体验,或其它业务问题。
  • 配置参考:
spring:kafka:producer:bootstrap-servers: 10.1.1.1:9092key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      retries:10000properties:delivery.timeout.ms:2000# 发送消息上报成功或失败的最大时间,默认120000,两分钟linger.ms:0# 生产者把数据组合到一个批处理进行请求的最大延迟时间,默认0# 参考 https://cwiki.apache.org/confluence/display/KAFKA/KIP-19+-+Add+a+request+timeout+to+NetworkClientrequest.timeout.ms:1000# 批处理就绪后到响应的等待时长,含网络+服务器复制时间batch.size:1000

2、消费者配置

  • 为了避免消息丢失,消费者需要开启手动ack,消息业务逻辑处理完成再提交偏移量
  • 参考后面的死循环问题,建议使用String反序列化器
  • 根据业务情况,配置合适的批量拉取数量max-poll-records,默认值500
  • 根据业务情况,配置合适的auto-offset-reset值,默认值latest - latest:消费者在消费主题的某个分区时,如果没有之前的消费记录(以前提交的偏移量),则只拉取最新消息,忽略历史消息。- earliest:与latest相反,没有之前的消费记录时,从最早的消息开始处理。- none:没有之前的消费记录时,抛出异常。
  • 配置参考:
spring:kafka:consumer:bootstrap-servers: 10.1.1.1:9092max-poll-records:100key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: latest
    listener:type: batch
      ack-mode: manual_immediate

3、其它

  • 配置多个 bootstrap server url,避免单节点故障,导致连接失败
  • 如果是异步发消息,不要在kafkaTemplate的成功回调和失败回调方法里有太多业务逻辑,回调方法是单线程处理,里面的业务逻辑会占用delivery.timeout.ms的超时时间配置,可能导致后续消息发送超时。
  • 同理,消费者也是单线程处理,消费逻辑太重的话,可能导致session.timeout.ms超时,从而被认为消费者离线,导致问题

四、kafka工具介绍

1、图形化工具

推荐

OffsetExplorer

,下载地址:https://www.kafkatool.com/download.html
在这里插入图片描述

2、命令行工具

kafka安装包内置了很多脚本工具,可以方便的查询kafka的状态,这些工具只需要下载就可以使用,无需安装。

  • 下载地址: https://kafka.apache.org/downloads 下载后解压,在bin目录下有很多sh文件,这些是在linux上使用的; 如果在Windows下使用,要用 bin\windows\ 下的那些bat文件。 下面用windows的bat文件命令举例(linux改用对应的sh文件执行即可)
  • 使用说明,请参考官方文档:https://kafka.apache.org/documentation/

查询某个消费者组下有哪些消费者,以及这些消费者对主题的消费状态:

d:\kafka_2.13-3.4.0\bin\windows\kafka-consumer-groups.bat --describe --group=cb_consumers --bootstrap-server=10.0.0.1:9092

在这里插入图片描述
字段说明:

  • GROUP 消费者分组
  • TOPIC 消费的主题
  • PARTITION 消费的分区
  • CURRENT-OFFSET 当前消费到的消息偏移量
  • LOG-END-OFFSET 当前分区的最大消息偏移量
  • LAG 滞后消息条数
  • CONSUMER-ID 消费者的ID
  • HOST 消费者所在的主机
  • CLIENT-ID 客户端ID 注:LAG可以简单理解为 LOG-END-OFFSET 减 CURRENT-OFFSET,但是实际上LAG=HW 减 CURRENT-OFFSET

四、springboot项目使用

1、生产者Demo代码:

1.1、添加pom依赖:

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

1.2、添加application.yml配置:

spring:kafka:producer:bootstrap-servers: 10.1.1.1:9092key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      retries:2# 失败重发次数

1.3、Java发送代码:

privatefinalKafkaTemplate kafkaTemplate;// 注入的Bean// 同步发送消息String topic ="beinetTest111";Object result = kafkaTemplate.send(topic,"我是key", objData).get();

2、消费者Demo代码:

2.1、添加pom依赖:

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

2.2、添加application.yml配置:

spring:kafka:consumer:bootstrap-servers: 10.1.1.1:9092key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer 
    listener:type: batch
      ack-mode: manual_immediate

2.3、Java发送代码:

@KafkaListener(topics ="${kafka-topic.reports}")publicvoidconsumerCreateTask(List<ConsumerRecord<String,Object>> consumerRecordList,Acknowledgment ack){if(consumerRecordList ==null|| consumerRecordList.size()<=0)return;long start =System.nanoTime();ConsumerRecord lastRecord = consumerRecordList.get(0);try{// 转换dto,并进行业务逻辑处理long elapsedTime =System.nanoTime()- start;
        log.debug("Topic:{} 分区:{} 偏移:{} 条数:{} 耗时:{}ns",
                lastRecord.topic(),
                lastRecord.partition(),
                lastRecord.offset(),
                dtos.size(),
                elapsedTime);}catch(Exception exp){long elapsedTime =System.nanoTime()- start;
        log.error("Topic:{} 分区:{} 偏移:{} 耗时:{}ns 出错:",
                lastRecord.topic(),
                lastRecord.partition(),
                lastRecord.offset(),
                elapsedTime,
                exp);}finally{// 不论成败,都提交,避免出错导致死循环,避免丢消息的逻辑,可以在catch里备份
        ack.acknowledge();}}

五、kafka常见问题

1、有多个消费者,但是总会有一个消费者拿不到消息数据

对一个消费者组而言,一个主题有几个分区,就最多接受几个消费者;
比如主题有2个分区,那么每个分区只能分配给组里的一个消费者,最多只有2个消费者连接上来,如果组里有3个消费者,那么肯定会有一个消费者处于空闲状态,没活干。
如果主题有2个分区,但是组里只有一个消费者,那么2个分区的消息,都会投递给这一个消费者。

2、主题的分区分配策略是怎么样的?

当主题存在多个分区和多个消费者时,kafka的源码实现里有如下几种分区分配策略:

  • Range策略(默认策略): 把当前消费者组消费的每个主题的所有分区,逐个分配给消费者,注意是每个主题单独处理,所以会出现不均衡的情况。 例如:a主题有3个分区a0/a1/a2,b主题3个分区b0/b1/b2,有2个消费者C0/C1,分配过程大致是: 第1步分配a主题: a0->C0, a1->C1, a2->C0 第2步分配b主题: b0->C0, b1->C1, b2->C0 这样可以看出,【消费者C0要维护4个分区的数据,而C1只要维护2个分区的数据】,出现了明显的不均衡问题。
  • Round-Robin策略: 把所有的分区,排序后,轮询方式逐一分配给所有的消费者, 例如:a主题有3个分区a0/a1/a2,b主题3个分区b0/b1/b2,有2个消费者C0/C1,分配过程大致是: 第1步分配a主题: a0->C0, a1->C1, a2->C0 第2步分配b主题: b0->C1, b1->C0, b2->C1 注意:第2步不是从头开始,而是接着第1步,继续分配,所以排除了Range方案的不均衡问题, 最终的分配结果是【2个消费者,各自负责3个分区】。 但是,如果2个消费者消费的主题,只有部分交集,并不完全相同时,还是会出现不均衡的情况。 如果希望改用这种策略,目前暂时没有配置方法,要在代码里修改partition.assignment.strategy属性,参考代码:
@Configuration@RequiredArgsConstructorpublicclassKafkaConfiguration{privatefinalKafkaProperties kafkaProperties;privatefinalConcurrentKafkaListenerContainerFactory<String,Object> kafkaFactory;@Bean("myKafkaFactory")publicKafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String,Object>>batchFactory(){Map<String,Object> props = kafkaProperties.buildConsumerProperties();
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
        kafkaFactory.setConsumerFactory(newDefaultKafkaConsumerFactory<>(props));return kafkaFactory;}}

然后在消费者代码上指定使用这个工厂Bean:

@KafkaListener(id ="beinetHandler1", groupId ="beinetGroup", topicPattern ="beinetTest.*",
            containerFactory ="myKafkaFactory")publicvoidmsgHandler(List<ConsumerRecord> message,Acknowledgment ack){
  • 在kafka的源码里还有2种实现,本文暂不深入介绍:org.apache.kafka.clients.consumer.CooperativeStickyAssignor``````org.apache.kafka.clients.consumer.StickyAssignor

3、消费者加入或退出时,消息还能正常消费吗?

结论:只要有存活的消费者存在,那么所有的消息都能正常消费。
当有新的消费者加入组,或组中有消费者下线/退出,都会触发消费者重新平衡的动作,就是重新为所有的消费者分配分区。
重平衡发生时,默认停止所有消费者工作,直到分配结束。

4、有兄弟说已经往kafka写入消息了,但是消费者那边没有数据入库

  • 首先,确认kafka有消息,用上面的图形化工具offset explorer,去对应的topic主题查找数据,发现确实有数据
  • 再在工具里,查看Consumers下的对应Group,发现Lag为0,说明消息已经被正常消费了
  • 查看消费者的应用日志,没有消费日志产生
  • 再继续排查消费者的应用日志,发现有如下日志:cb_consumers: partitions assigned: [] 这表示,该消费者没有分配到分区,不在工作中。 初步判断,是不是有人在其它地方启动了消费者,把数据给消费掉了。
  • offset explorer这个工具,不能展示消费者IP信息,只能使用上面的命令kafka-consumer-groups.bat查看消费者IP, 再找运维看看这个IP是谁的。
  • 最后定位到是测试环境配置错误,把开发环境的数据消费掉了。

5、反序列化失败,导致消费者死循环问题

某天发布到测试环境后,发现程序启动后,一直抛如下异常,且持续几十分钟也不会中断:

<#6d8d6458> j.l.IllegalStateException: No type information in headers and no default type provided
    at o.s.util.Assert.state(Assert.java:76)
    at o.s.k.s.s.JsonDeserializer.deserialize(JsonDeserializer.java:535)
    at o.a.k.c.c.i.Fetcher.parseRecord(Fetcher.java:1387)
    at o.a.k.c.c.i.Fetcher.access$3400(Fetcher.java:133)
    at o.a.k.c.c.i.Fetcher$CompletedFetch.fetchRecords(Fetcher.java:1618)
    at o.a.k.c.c.i.Fetcher$CompletedFetch.access$1700(Fetcher.java:1454)
    at o.a.k.c.c.i.Fetcher.fetchRecords(Fetcher.java:687)
    at o.a.k.c.c.i.Fetcher.fetchedRecords(Fetcher.java:638)
    at o.a.k.c.c.KafkaConsumer.pollForFetches(KafkaConsumer.java:1272)
    at o.a.k.c.c.KafkaConsumer.poll(KafkaConsumer.java:1233)
    at o.a.k.c.c.KafkaConsumer.poll(KafkaConsumer.java:1206)
    at j.i.r.GeneratedMethodAccessor109.invoke(Unknown Source)
    at j.i.r.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at j.l.reflect.Method.invoke(Unknown Source)
    at o.s.a.s.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
    at o.s.a.f.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:208)
    at c.s.proxy.$Proxy186.poll(Unknown Source)
    at o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1413)
    at o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1250)
    at o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1162)
    at j.u.c.Executors$RunnableAdapter.call(Unknown Source)
    at j.u.c.FutureTask.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)

放google搜索了一下,说是反序列化找不到类型信息导致的,并建议不要使用JSON反序列化。
查了一下配置变化记录,确实加了一个kafka的反序列化配置变更:

spring:kafka:producer:value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer

把spring.consumer.value-deserializer 改成:

org.apache.kafka.common.serialization.StringDeserializer

就恢复了。
了解了一下,同事希望在消费者的方法参数上,直接使用对象,而不是使用String,所以做了这个修改。
而正好出错的这个消费者,消费的是其它项目生产的消息,里面并不包含类型信息。

而这个异常,是在spring的底层抛出的,业务代码上无法进行try 捕捉,代码同时又设置了手工提交ack,导致代码进入了死循环。
为了避免这种问题,还是建议使用StringDeserializer反序列化,自己在代码里反序列化比较好。

6、broker单个故障,导致消费者无法提交偏移量的问题

生产环境,为了性能和故障转移,部署了6个broker,某天有一个broker故障下线了,按理应该会自动切换,实际上,所有的消费者都开始抛异常:“error when storing group assignment during syncgroup”
直到人工恢复broker并上线,故障才恢复。
最终排查结果:

  • 运维把kafka的一个内部topic: __consumer_offsets 副本数配置为2,
  • 同时配置 min.insync.replicas=2 ,该配置的含义是ISR列表最小同步副本数不得少于2个
  • 故障下线的broker,正好包含该topic: __consumer_offsets的一个副本,导致该主题的副本数只剩下一个,不符合 min.insync.replicas=2配置要求,从而停止工作
  • topic: __consumer_offsets的作用,是接收并存储所有消费者组的消费偏移量,该主题不工作,就会导致消费者无法提交偏移量,从而导致所有消费不正常,会重复消费数据。

知道问题了,调整就是把 topic:

__consumer_offsets

的副本数调整为3(默认值就是3,运维改错了)

7、所有消费者都不消费任何消息

如果消费者先启动,然后才创建topic,会导致消费者消费不到数据,可以重启消费者试试

8、spring中的kafka,是否存在线程安全问题

生产者使用的KafkaTemplate是线程安全的,经测试,都是使用同一个线程进行消息发送。
同样,消费者也是线程安全的,每个消费者也是单线程处理所有接收到的消息。

9、kafka消息堵塞怎么处理

  • 如果消息不重要,可以直接删除主题重建,该主题下的所有消息自然就没了,注意需要重建topic,然后重启消费者;
  • 如果所有消息都需要消费,基于kafka的分区特性,每个分区只能一个消费者,因此无法简单通过增加消费者来解决问题 - 确认消费者有没有出现异常,需要注意的是,有些开发人员会吞掉异常,导致你认为消费者正常的,可以通过业务数据是否持续增长来判别,如果是消费者异常了,修复bug即可。- 如果消费正常,再确认是否突发消息数据增长,简单判断就是主题的LAG是否在持续按正常速率降低,观察几分钟,如果持续降低,基于判断是突发消息,可以耐心等待。- 如果消息不会正常下降,基本判断是消费速度慢, - 先用工具确认该主题下每个分区的LAG,如果只是某个分区的LAG特别高,其它分区正常(未堵塞),那么应该是消息分配不均衡,这个分区的消息特别多,考虑调整生产者的消息key,确保所有分区的消息数量均衡;- 如果对消息时序无特别要求,可以代码里通过线程池异步处理消息,注意要控制线程池数量,不要导致应用oom了;- 考虑增加几个分区,再增加几个消费者,这样新生产的消息会重新分散到不同分区,降低旧消费者的压力;- 堵塞的消息,考虑加个新消费者,消费到另外的临时主题,临时主题多加个几倍的分区和消费者,进行快速消费,注意不要对下游或数据库造成冲击,导致其它问题。
标签: kafka spring boot

本文转载自: https://blog.csdn.net/youbl/article/details/130286729
版权归原作者 游北亮 所有, 如有侵权,请联系我们删除。

“kafka使用详解、最佳实践和问题排查”的评论:

还没有评论