目录

• 1. 整合kafka
• 2. 消息发送
• 2.1 发送类型
• 2.2 序列化
• 2.3 分区策略
• 3. 消息消费
• 3.1 消息组别
• 3.2 位移提交

【Springboot整合kafka的示例代码】
1. 整合kafka 1、引入依赖

org.springframework.kafkaspring-kafka

2、设置yml文件

spring:application:name: demokafka:bootstrap-servers: 52.82.98.209:10903,52.82.98.209:10904producer: # 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: com.itheima.demo.config.MySerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer: # consumer消费者group-id: javagroup # 默认的消费组IDenable-auto-commit: true # 是否自动提交offsetauto-commit-interval: 100# 提交offset延时(接收到消息后多久提交offset)# earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费# latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据# none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常auto-offset-reset: latestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializer#value-deserializer: com.itheima.demo.config.MyDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

3、启动项目

文章图片


2. 消息发送
2.1 发送类型
KafkaTemplate调用send时默认采用异步发送，如果需要同步获取发送结果，调用get方法
异步发送生产者：

@RestControllerpublic class KafkaProducer {@Resourceprivate KafkaTemplate kafkaTemplate; @GetMapping("/kafka/test/{msg}")public void sendMessage(@PathVariable("msg") String msg) {Message message = new Message(); message.setMessage(msg); kafkaTemplate.send("test", JSON.toJSONString(message)); }}

同步发送生产者：

//测试同步发送与监听@RestControllerpublic class AsyncProducer {private final static Logger logger = LoggerFactory.getLogger(AsyncProducer.class); @Resourceprivate KafkaTemplate kafkaTemplate; //同步发送@GetMapping("/kafka/sync/{msg}")public void sync(@PathVariable("msg") String msg) throws Exception {Message message = new Message(); message.setMessage(msg); ListenableFuture> future = kafkaTemplate.send("test", JSON.toJSONString(message)); //注意，可以设置等待时间，超出后，不再等候结果SendResult result = future.get(3,TimeUnit.SECONDS); logger.info("send result:{}",result.getProducerRecord().value()); }}

消费者：

@Componentpublic class KafkaConsumer {private final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class); //不指定group，默认取yml里配置的@KafkaListener(topics = {"test"})public void onMessage1(ConsumerRecord consumerRecord) {Optional optional = Optional.ofNullable(consumerRecord.value()); if (optional.isPresent()) {Object msg = optional.get(); logger.info("message:{}", msg); }}}

那么我们怎么看出来同步发送和异步发送的区别呢？
①首先在服务器上，将kafka暂停服务。
②在swagger发送消息
调同步发送：请求被阻断，一直等待，超时后返回错误

文章图片

而调异步发送的（默认发送接口），请求立刻返回。

文章图片

那么，异步发送的消息怎么确认发送情况呢？
我们使用注册监听
即新建一个类：KafkaListener.java

@Configurationpublic class KafkaListener {private final static Logger logger = LoggerFactory.getLogger(KafkaListener.class); @AutowiredKafkaTemplate kafkaTemplate; //配置监听@PostConstructprivate void listener() {kafkaTemplate.setProducerListener(new ProducerListener() {@Overridepublic void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata) {logger.info("ok,message={}", producerRecord.value()); }public void onError(ProducerRecord producerRecord, Exception exception) {logger.error("error!message={}", producerRecord.value()); }); }}

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

文章图片

如果是正常发送异步消息，则会获得该消息。可以看到，在内部类 KafkaListener$1 中，即注册的Listener的消息。

文章图片


2.2 序列化
消费者使用：KafkaConsumer.java

@Componentpublic class KafkaConsumer {private final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class); //不指定group，默认取yml里配置的@KafkaListener(topics = {"test"})public void onMessage1(ConsumerRecord consumerRecord) {Optional optional = Optional.ofNullable(consumerRecord.value()); if (optional.isPresent()) {Object msg = optional.get(); logger.info("message:{}", msg); }}}

1）序列化详解

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

2）自定义序列化
自己实现，实现对应的接口即可，有以下方法：

public interface Serializer extends Closeable { default void configure(Map 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

public class MySerializer implements Serializer {@Overridepublic byte[] serialize(String s, Object o) {String json = JSON.toJSONString(o); return json.getBytes(); }}

3）解码
MyDeserializer.java，实现方式与编码器几乎一样.

public class MyDeserializer implements Deserializer {private final static Logger logger = LoggerFactory.getLogger(MyDeserializer.class); @Overridepublic Object deserialize(String s, byte[] bytes) {try {String json = new String(bytes,"utf-8"); return JSON.parse(json); } catch (UnsupportedEncodingException e) {e.printStackTrace(); }return null; }}

4）在yaml中配置自己的编码器、解码器

文章图片

再次收发，消息正常

文章图片


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

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

1）验证默认分区规则
发送者代码参考：PartitionProducer.java

//测试分区发送@RestControllerpublic class PartitionProducer {@Resourceprivate KafkaTemplate kafkaTemplate; //指定分区发送//不管你key是什么，到同一个分区@GetMapping("/kafka/partitionSend/{key}")public void setPartition(@PathVariable("key") String key) {kafkaTemplate.send("test", 0, key, "key=" + key + "，msg=指定0号分区"); }//指定key发送，不指定分区//根据key做hash，相同的key到同一个分区@GetMapping("/kafka/keysend/{key}")public void setKey(@PathVariable("key") String key) {kafkaTemplate.send("test", key, "key=" + key + "，msg=不指定分区"); }

消费者代码使用：PartitionConsumer.java

@Componentpublic class PartitionConsumer {private final Logger logger = LoggerFactory.getLogger(PartitionConsumer.class); //分区消费@KafkaListener(topics = {"test"},topicPattern = "0")public void onMessage(ConsumerRecord consumerRecord) {Optional optional = Optional.ofNullable(consumerRecord.value()); if (optional.isPresent()) {Object msg = optional.get(); logger.info("partition=0,message:[{}]", msg); }}@KafkaListener(topics = {"test"},topicPattern = "1")public void onMessage1(ConsumerRecord consumerRecord) {Optional optional = Optional.ofNullable(consumerRecord.value()); if (optional.isPresent()) {Object msg = optional.get(); logger.info("partition=1,message:[{}]", msg); }}}

通过swagger访问setKey(也就是只给了key的方法)：

文章图片

可以看到key相同的被hash到了同一个分区

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

文章图片

可以看到无论key是什么，都是分区0来消费

2）自定义分区
参考代码：MyPartitioner.java , MyPartitionTemplate.java。
发送使用：MyPartitionProducer.java。

public class MyPartitioner implements Partitioner {@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {//定义自己的分区策略//如果key以0开头，发到0号分区//其他都扔到1号分区String keyStr = key+""; if (keyStr.startsWith("0")){return 0; }else {return 1; }}public void close() {public void configure(Map map) {}

@Configurationpublic class MyPartitionTemplate { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @Value("${spring.kafka.bootstrap-servers}")private String bootstrapServers; KafkaTemplate kafkaTemplate; @PostConstructpublic void setKafkaTemplate() {Map props = new HashMap<>(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); //注意分区器在这里！！！props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class); this.kafkaTemplate = new KafkaTemplate(new DefaultKafkaProducerFactory<>(props)); }public KafkaTemplate getKafkaTemplate(){return kafkaTemplate; }

//测试自定义分区发送@RestControllerpublic class MyPartitionProducer {@AutowiredMyPartitionTemplate template; //使用0开头和其他任意字母开头的key发送消息//看控制台的输出，在哪个分区里？@GetMapping("/kafka/myPartitionSend/{key}")public void setPartition(@PathVariable("key") String key) {template.getKafkaTemplate().send("test", key,"key="+key+"，msg=自定义分区策略"); }}

使用swagger，发送0开头和非0开头两种key

文章图片


3. 消息消费
3.1 消息组别
发送者使用：KafkaProducer.java

@RestControllerpublic class KafkaProducer {@Resourceprivate KafkaTemplate kafkaTemplate; @GetMapping("/kafka/test/{msg}")public void sendMessage(@PathVariable("msg") String msg) {Message message = new Message(); message.setMessage(msg); kafkaTemplate.send("test", JSON.toJSONString(message)); }}

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

//测试组消费@Componentpublic class GroupConsumer {private final Logger logger = LoggerFactory.getLogger(GroupConsumer.class); //组1，消费者1@KafkaListener(topics = {"test"},groupId = "group1")public void onMessage1(ConsumerRecord consumerRecord) {Optional optional = Optional.ofNullable(consumerRecord.value()); if (optional.isPresent()) {Object msg = optional.get(); logger.info("group:group1-1 , message:{}", msg); }}//组1，消费者2public void onMessage2(ConsumerRecord consumerRecord) {logger.info("group:group1-2 , message:{}", msg); //组2，只有一个消费者@KafkaListener(topics = {"test"},groupId = "group2")public void onMessage3(ConsumerRecord consumerRecord) {logger.info("group:group2 , message:{}", msg); }

2）启动

文章图片

3）通过swagger发送2条消息

文章图片

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

4）消费端闲置
注意分区数与消费者数的搭配，如果 （ 消费者数 > 分区数量 ），将会出现消费者闲置（因为一个分区只能分配给一个消费者），浪费资源！
验证方式：
停掉项目，删掉test主题，重新建一个 ，这次只给它分配一个分区。
重新发送两条消息，试一试

文章图片

• group2可以消费到1、2两条消息
• group1下有两个消费者，但是只分配给了 1 ， 2这个进程被闲置

3.2 位移提交
1）自动提交
前面的案例中，我们设置了以下两个选项，则kafka会按延时设置自动提交

enable-auto-commit: true # 是否自动提交offsetauto-commit-interval: 100 # 提交offset延时(接收到消息后多久提交offset，默认单位为ms)

2）手动提交
有些时候，我们需要手动控制偏移量的提交时机，比如确保消息严格消费后再提交，以防止丢失或重复。
下面我们自己定义配置，覆盖上面的参数
代码参考：MyOffsetConfig.java

@Configurationpublic class MyOffsetConfig {private final Logger logger = LoggerFactory.getLogger(this.getClass()); @Value("${spring.kafka.bootstrap-servers}")private String bootstrapServers; @Beanpublic KafkaListenerContainerFactory manualKafkaListenerContainerFactory() {Map configProps = new HashMap<>(); configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); // 注意这里！！！设置手动提交configProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); ConcurrentKafkaListenerContainerFactory factory =new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(configProps)); // ack模式：//AckMode针对ENABLE_AUTO_COMMIT_CONFIG=false时生效，有以下几种：////RECORD//每处理一条commit一次//BATCH(默认)//每次poll的时候批量提交一次，频率取决于每次poll的调用频率//TIME//每次间隔ackTime的时间去commit(跟auto commit interval有什么区别呢？)//COUNT//累积达到ackCount次的ack去commit//COUNT_TIME//ackTime或ackCount哪个条件先满足，就commit//MANUAL//listener负责ack，但是背后也是批量上去//MANUAL_IMMEDIATE//listner负责ack，每调用一次，就立即commitfactory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); return factory; }}

然后通过在消费端的Consumer来提交偏移量
MyOffsetConsumer：

@Componentpublic class MyOffsetConsumer {private final Logger logger = LoggerFactory.getLogger(this.getClass()); @KafkaListener(topics = "test", groupId = "myoffset-group-1", containerFactory = "manualKafkaListenerContainerFactory")public void manualCommit(@Payload String message,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,@Header(KafkaHeaders.RECEIVED_TOPIC) String topic,Consumer consumer,Acknowledgment ack) {logger.info("手动提交偏移量 , partition={}, msg={}", partition, message); // 同步提交consumer.commitSync(); //异步提交//consumer.commitAsync(); // ack提交也可以，会按设置的ack策略走(参考MyOffsetConfig.java里的ack模式)// ack.acknowledge(); }@KafkaListener(topics = "test", groupId = "myoffset-group-2", containerFactory = "manualKafkaListenerContainerFactory")public void noCommit(@Payload String message,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,@Header(KafkaHeaders.RECEIVED_TOPIC) String topic,Consumer consumer,Acknowledgment ack) {logger.info("忘记提交偏移量, partition={}, msg={}", partition, message); // 不做commit！/*** 现实状况：* commitSync和commitAsync组合使用* * 手工提交异步 consumer.commitAsync(); * 手工同步提交 consumer.commitSync()* commitSync()方法提交最后一个偏移量。在成功提交或碰到无怯恢复的错误之前，* commitSync()会一直重试，但是commitAsync()不会。* 一般情况下，针对偶尔出现的提交失败，不进行重试不会有太大问题* 因为如果提交失败是因为临时问题导致的，那么后续的提交总会有成功的。* 但如果这是发生在关闭消费者或再均衡前的最后一次提交，就要确保能够提交成功。否则就会造成重复消费* 因此，在消费者关闭前一般会组合使用commitAsync()和commitSync()。*///@KafkaListener(topics = "test", groupId = "myoffset-group-3",containerFactory = "manualKafkaListenerContainerFactory")public void manualOffset(@Payload String message,try {logger.info("同步异步搭配 , partition={}, msg={}", partition, message); //先异步提交consumer.commitAsync(); //继续做别的事} catch (Exception e) {System.out.println("commit failed"); } finally {try {consumer.commitSync(); } finally {consumer.close(); }}* 甚至可以手动提交，指定任意位置的偏移量* 不推荐日常使用！！！//@KafkaListener(topics = "test", groupId = "myoffset-group-4",containerFactory = "manualKafkaListenerContainerFactory")public void offset(ConsumerRecord record, Consumer consumer) {logger.info("手动指定任意偏移量， partition={}, msg={}", record.partition(), record); Map currentOffset = new HashMap<>(); currentOffset.put(new TopicPartition(record.topic(), record.partition()),new OffsetAndMetadata(record.offset() + 1)); consumer.commitSync(currentOffset); }

3）重复消费问题
如果手动提交模式被打开，一定不要忘记提交偏移量。否则会造成重复消费！
用km将test主题删除，新建一个test空主题。方便观察消息偏移 注释掉其他Consumer的Component注解，只保留当前MyOffsetConsumer.java 启动项目，使用swagger的KafkaProducer发送连续几条消息 留心控制台，都能消费，没问题：

文章图片

但是！重启项目：

文章图片

无论重启多少次，不提交偏移量的消费组，会重复消费一遍！！！
再通过命令行查询偏移量

文章图片

4）经验与总结
commitSync()方法，即同步提交，会提交最后一个偏移量。在成功提交或碰到无怯恢复的错误之前，commitSync()会一直重试，但是commitAsync()不会。
这就造成一个陷阱：
如果异步提交，针对偶尔出现的提交失败，不进行重试不会有太大问题，因为如果提交失败是因为临时问题导致的，那么后续的提交总会有成功的。只要成功一次，偏移量就会提交上去。
但是！如果这是发生在关闭消费者时的最后一次提交，就要确保能够提交成功，如果还没提交完就停掉了进程。就会造成重复消费！
因此，在消费者关闭前一般会组合使用commitAsync()和commitSync()。
详细代码参考：MyOffsetConsumer.manualOffset()
到此这篇关于Springboot整合kafka的文章就介绍到这了,更多相关Springboot整合kafka内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

推荐阅读

• 

  dnf20级魔法秀数据_dnf魔法秀是多少级技能
• 

  公积金怎么取出来装修 公积金如何取出来装修
• 

  冠状动脉支架|中国广泛存在冠状动脉支架滥用的结论或过于武断
• 

  心肌炎会不会传染？ 心肌炎传染吗
• 

  买冰箱去哪？如何选择适合自己的冰箱？
• 

  为什么被子晒了有股味道 被子晒了有股味道的原因
• 

  因病休学申请书
• 

  孕妇能不能用花露水喷腿 孕妇能不能用花露水六神
• 

  材质分析与布光,这个杀手不太冷布光分析
• 

  adolbook是什么牌子
• 

  产品概述
• 

  王者荣耀的王者时刻支持哪些机型 王者时刻支持哪些机型2020
• 

  米家电动牙刷
• 

  三菱重工空调显示EP5怎么处理,快来了解一下
• 

  到机场后怎么找航站楼接人 到机场后怎么找航站楼
• 

  没系安全带扣几分 不同的公路上开车扣分和罚款是不同的
• 

  水晶玉露的养殖方法 水晶玉露如何养殖
• 

  关于超频的概念
• 

  社保|南昌：企业改制裁员，职工发现社保漏缴，负责人：我没办法解释
• 

  Android实现缓存大图到SD卡

• Activiti（一）SpringBoot2集成Activiti6
• SpringBoot调用公共模块的自定义注解失效的解决
• 解决SpringBoot引用别的模块无法注入的问题
• 深入浅出谈一下有关分布式消息技术（Kafka）
• Spring|Spring Boot 整合 Activiti6.0.0
• springboot使用redis缓存
• 15.Kafka
• springboot整合数据库连接池-->druid
• SpringBoot中YAML语法及几个注意点说明
• springboot结合redis实现搜索栏热搜功能及文字过滤