Springboot集成kafka高级应用实战分享

目录
  • 深入应用
    • 1.1 springboot-kafka
    • 1.2 消息发送
      • 1.2.1 发送类型
      • 1.2.2 序列化
      • 1.2.3 分区策略
    • 1.3 消息消费
      • 1.3.1 消息组别
      • 1.3.2 位移提交

深入应用

1.1 springboot-kafka

1)配置文件

kafka:
bootstrap-servers: 52.82.98.209:10903,52.82.98.209:10904
producer: # producer 生产者
retries: 0 # 重试次数
acks: 1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
batch-size: 16384 # 一次最多发送数据量
buffer-memory: 33554432 # 生产端缓冲区大小
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer

consumer: # consumer消费者
group-id: javagroup # 默认的消费组ID
enable-auto-commit: true # 是否自动提交offset
auto-commit-interval: 100 # 提交offset延时(接收到消息后多久提交offset)
auto-offset-reset: latest #earliest,latest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

2)启动信息

1.2 消息发送

1.2.1 发送类型

KafkaTemplate调用send时默认采用异步发送,如果需要同步获取发送结果,调用get方法

详细代码参考:AsyncProducer.java

消费者使用:KafkaConsumer.java

1)同步发送

ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send("test", JSON.toJSONString(message));
//注意,可以设置等待时间,超出后,不再等候结果
SendResult<String, Object> result = future.get(3,TimeUnit.SECONDS);
logger.info("send result:{}",result.getProducerRecord().value());

通过swagger发送,控制台可以正常打印send result

2)阻断

在服务器上,将kafka暂停服务

docker-compose pause kafka-1 kafka-2

在swagger发送消息

调同步发送:请求被阻断,一直等待,超时后返回错误

而调异步发送的(默认发送接口),请求立刻返回。

那么,异步发送的消息怎么确认发送情况呢???往下看!

3)注册监听

代码参考: KafkaListener.java

可以给kafkaTemplate设置Listener来监听消息发送情况,实现内部的对应方法

kafkaTemplate.setProducerListener(new ProducerListener<String, Object>() {});

查看控制台,等待一段时间后,异步发送失败的消息会被回调给注册过的listener

com.itheima.demo.config.KafkaListener:error!message={"message":"1","sendTime":1609920296374}

启动kafka

docker-compose unpause kafka-1 kafka-2

再次发送消息时,同步异步均可以正常收发,并且监听进入success回调

com.itheima.demo.config.KafkaListener$1:ok,message={"message":"1","sendTime":1610089315395}
com.itheima.demo.controller.PartitionConsumer:patition=1,message:[{"message":"1","sendTime":1610089315395}]

可以看到,在内部类 KafkaListener$1 中,即注册的Listener的消息。

1.2.2 序列化

消费者使用:KafkaConsumer.java

1)序列化详解

  • 前面用到的是Kafka自带的字符串序列化器(org.apache.kafka.common.serialization.StringSerializer)
  • 除此之外还有:ByteArray、ByteBuffer、Bytes、Double、Integer、Long 等
  • 这些序列化器都实现了接口 (org.apache.kafka.common.serialization.Serializer)
  • 基本上,可以满足绝大多数场景

2)自定义序列化

自己实现,实现对应的接口即可,有以下方法:

public interface Serializer<T> extends Closeable {
default void configure(Map<String, ?> configs, boolean isKey) {
}

//理论上,只实现这个即可正常运行
byte[] serialize(String var1, T var2);

//默认调上面的方法
default byte[] serialize(String topic, Headers headers, T data) {
return this.serialize(topic, data);
}

default void close() {
}
}

案例,参考: MySerializer.java

在yaml中配置自己的编码器

value-serializer: com.itheima.demo.config.MySerializer

重新发送,发现:消息发送端编码回调一切正常。但是消费端消息内容不对!

com.itheima.demo.controller.KafkaListener$1:ok,message={"message":"1","sendTime":1609923570477}
com.itheima.demo.controller.KafkaConsumer:message:"{\"message\":\"1\",\"sendTime\":1609923570477}"

怎么办?

3)解码

发送端有编码并且我们自己定义了编码,那么接收端自然要配备对应的解码策略

代码参考:MyDeserializer.java,实现方式与编码器几乎一样!

在yaml中配置自己的解码器

value-deserializer: com.itheima.demo.config.MyDeserializer

再次收发,消息正常

com.itheima.demo.controller.AsyncProducer$1:ok,message={"message":"1","sendTime":1609924855896}
com.itheima.demo.controller.KafkaConsumer:message:{"message":"1","sendTime":1609924855896}

1.2.3 分区策略

分区策略决定了消息根据key投放到哪个分区,也是顺序消费保障的基石。

  • 给定了分区号,直接将数据发送到指定的分区里面去
  • 没有给定分区号,给定数据的key值,通过key取上hashCode进行分区
  • 既没有给定分区号,也没有给定key值,直接轮循进行分区
  • 自定义分区,你想怎么做就怎么做

1)验证默认分区规则

发送者代码参考:PartitionProducer.java

消费者代码使用:PartitionConsumer.java

通过swagger访问setKey:

看控制台:

再访问setPartition来设置分区号0来发送

看控制台:

2)自定义分区

你想自己定义规则,根据我的要求,把消息投放到对应的分区去? 可以!

参考代码:MyPartitioner.java , MyPartitionTemplate.java ,

发送使用:MyPartitionProducer.java

使用swagger,发送0开头和非0开头两种key试一试!

备注:

自己定义config参数,比较麻烦,需要打破默认的KafkaTemplate设置

可以将KafkaConfiguration.java中的getTemplate加上@Bean注解来覆盖系统默认bean

这里为了避免混淆,采用@Autowire注入

1.3 消息消费

1.3.1 消息组别

发送者使用:KafkaProducer.java

1)代码参考:GroupConsumer.java,Listener拷贝3份,分别赋予两组group,验证分组消费:

2)启动

3)通过swagger发送2条消息

  • 同一group下的两个消费者,在group1均分消息
  • group2下只有一个消费者,得到全部消息

4)消费端闲置

注意分区数与消费者数的搭配,如果 ( 消费者数 > 分区数量 ),将会出现消费者闲置,浪费资源!

验证方式:

停掉项目,删掉test主题,重新建一个 ,这次只给它分配一个分区。

重新发送两条消息,试一试

解析:

group2可以消费到1、2两条消息

group1下有两个消费者,但是只分配给了 -1 , -2这个进程被闲置

1.3.2 位移提交

1)自动提交

前面的案例中,我们设置了以下两个选项,则kafka会按延时设置自动提交

enable-auto-commit: true # 是否自动提交offset
auto-commit-interval: 100 # 提交offset延时(接收到消息后多久提交offset)

2)手动提交

有些时候,我们需要手动控制偏移量的提交时机,比如确保消息严格消费后再提交,以防止丢失或重复。

下面我们自己定义配置,覆盖上面的参数

代码参考:MyOffsetConfig.java

通过在消费端的Consumer来提交偏移量,有如下几种方式:

代码参考:MyOffsetConsumer.java

同步提交、异步提交:manualCommit() ,同步异步的差别,下面会详细讲到。

指定偏移量提交:offset()

3)重复消费问题

如果手动提交模式被打开,一定不要忘记提交偏移量。否则会造成重复消费!

代码参考和对比:manualCommit() , noCommit()

验证过程:

用km将test主题删除,新建一个test空主题。方便观察消息偏移 注释掉其他Consumer的Component注解,只保留当前MyOffsetConsumer.java 启动项目,使用swagger的KafkaProducer发送连续几条消息 留心控制台,都能消费,没问题:

但是!重启试试:

无论重启多少次,不提交偏移量的消费组,会重复消费一遍!!!

再通过命令行查询偏移量试试:

4)经验与总结

commitSync()方法,即同步提交,会提交最后一个偏移量。在成功提交或碰到无怯恢复的错误之前,commitSync()会一直重试,但是commitAsync()不会。

这就造成一个陷阱:

如果异步提交,针对偶尔出现的提交失败,不进行重试不会有太大问题,因为如果提交失败是因为临时问题导致的,那么后续的提交总会有成功的。只要成功一次,偏移量就会提交上去。
但是!如果这是发生在关闭消费者时的最后一次提交,就要确保能够提交成功,如果还没提交完就停掉了进程。就会造成重复消费!
因此,在消费者关闭前一般会组合使用commitAsync()和commitSync()。

到此这篇关于Springboot集成kafka高级应用实战分享的文章就介绍到这了,更多相关Springboot集成kafka 内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • SpringBoot集成kafka全面实战记录

    本文是SpringBoot+Kafka的实战讲解,如果对kafka的架构原理还不了解的读者,建议先看一下<大白话kafka架构原理>.<秒懂kafka HA(高可用)>两篇文章. 一.生产者实践 普通生产者 带回调的生产者 自定义分区器 kafka事务提交 二.消费者实践 简单消费 指定topic.partition.offset消费 批量消费 监听异常处理器 消息过滤器 消息转发 定时启动/停止监听器 一.前戏 1.在项目中连接kafka,因为是外网,首先要开放kafka配置文件

  • Springboot集成Kafka进行批量消费及踩坑点

    目录 引入依赖 创建配置类 Kafka 消费者 引入依赖 <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>1.3.11.RELEASE</version> </dependency> 因为我的项目的 springboot 版本是 1.5.22.RELE

  • Springboot 2.x集成kafka 2.2.0的示例代码

    目录 引言 基本环境 代码编写 1.基本引用pom 2.基本配置 3.实体类 4.生产者端 5.消费者 6.测试 效果展示 遇到的问题 引言 kafka近几年更新非常快,也可以看出kafka在企业中是用的频率越来越高,在springboot中集成kafka还是比较简单的,但是应该注意使用的版本和kafka中基本配置,这个地方需要信心,防止进入坑中. 版本对应地址:https://spring.io/projects/spring-kafka 基本环境 springboot版本2.1.4 kafk

  • SpringBoot集成Kafka的步骤

    SpringBoot集成Kafka 本篇主要讲解SpringBoot 如何集成Kafka ,并且简单的 编写了一个Demo 来测试 发送和消费功能 前言 选择的版本如下: springboot : 2.3.4.RELEASE spring-kafka : 2.5.6.RELEASE kafka : 2.5.1 zookeeper : 3.4.14 本Demo 使用的是 SpringBoot 比较高的版本 SpringBoot 2.3.4.RELEASE 它会引入 spring-kafka 2.5

  • Springboot集成Kafka实现producer和consumer的示例代码

    本文介绍如何在springboot项目中集成kafka收发message. Kafka是一种高吞吐量的分布式发布订阅消息系统,有如下特性: 通过O(1)的磁盘数据结构提供消息的持久化,这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能.高吞吐量:即使是非常普通的硬件Kafka也可以支持每秒数百万的消息.支持通过Kafka服务器和消费机集群来分区消息.支持Hadoop并行数据加载. 安装Kafka 因为安装kafka需要zookeeper的支持,所以Windows安装时需要将zookee

  • springboot 1.5.2 集成kafka的简单例子

    本文介绍了springboot 1.5.2 集成kafka的简单例子 ,分享给大家,具体如下: 随着spring boot 1.5版本的发布,在spring项目中与kafka集成更为简便. 添加依赖 compile("org.springframework.kafka:spring-kafka:1.1.2.RELEASE") 添加application.properties #kafka # 指定kafka 代理地址,可以多个 spring.kafka.bootstrap-server

  • Springboot集成kafka高级应用实战分享

    目录 深入应用 1.1 springboot-kafka 1.2 消息发送 1.2.1 发送类型 1.2.2 序列化 1.2.3 分区策略 1.3 消息消费 1.3.1 消息组别 1.3.2 位移提交 深入应用 1.1 springboot-kafka 1)配置文件 kafka: bootstrap-servers: 52.82.98.209:10903,52.82.98.209:10904 producer: # producer 生产者 retries: 0 # 重试次数 acks: 1 #

  • springboot集成junit编写单元测试实战

    目录 一:查看jar包版本号是否为junit4: 二:实战应用: 三:扩展 在做单元测试时,代码覆盖率常常被拿来作为衡量测试好坏的指标,甚至,用代码覆盖率来考核测试任务完成情况,比如,代码覆盖率必须达到80%或 90%.于是乎,测试人员费尽心思设计案例覆盖代码.用代码覆盖率来衡量,有利也有弊. 首先,让我们先来了解一下所谓的“代码覆盖率”.我找来了所谓的定义:代码覆盖率 = 代码的覆盖程度,一种度量方式. 一:查看jar包版本号是否为junit4: junit自身注解: @BeforeClass

  • SpringBoot集成Kafka 配置工具类的详细代码

    目录 1.单播模式,只有一个消费者组 2.广播模式,多个消费者组 spring-kafka 是基于 java版的 kafka client与spring的集成,提供了 KafkaTemplate,封装了各种方法,方便操作,它封装了apache的kafka-client,不需要再导入client依赖 <!-- kafka --> <dependency> <groupId>org.springframework.kafka</groupId> <arti

  • SpringBoot整合RabbitMQ及生产全场景高级特性实战

    目录 摘要 整合 依赖与配置 生产者配置消息队列规则 生产者发布消息 消费者监听消息 摘要 整合场景含 topic 工作模式(通过 routingKey 可满足简单/工作队列/发布订阅/路由等四种工作模式)和 confirm(消息确认).return(消息返回).basicAck(消息签收).basicNack(拒绝签收).DLX(Dead Letter Exchange死信队列)实现延时/定时任务等. 整合 依赖与配置 以下内容消费者同生产者 <parent> <groupId>

随机推荐