0
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使用详解、最佳实践和问题排查”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

Docker完整卸载的两种方法详细教程例子解析

spark-libFM 项目常见问题解决方案

基于大数据 Python NBA球员数据分析系统(源码+LW+部署讲解+数据库+ppt)

软件测试笔记｜web自动化测试｜Web 自动化测试中，XPath 定位的方法有哪些？如何编写 XPath？

Spring Boot 日志打印配置详解

MySQL中日期和时间戳的转换：字符到DATE和TIMESTAMP的相互转换

VLAN高级+以太网安全

文章导航