devbox
Guff_9hys
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

SpringBoot整合kafka(实现producer和consumer)_spring.kafka.producer.acks

作者：Guff_9hys | 2024-07-23 14:36:00

赞

踩

spring.kafka.producer.acks

https://www.jianshu.com/p/5da86afed228

在Windows环境下安装运行Kafka：https://www.jianshu.com/p/d64798e81f3b

本文代码使用的是Spring Boot 2.1.1.RELEASE 版本

 

  1. <parent>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-parent</artifactId>
  4.     <version>2.1.1.RELEASE</version>
  5.     <relativePath/> <!-- lookup parent from repository -->
  6. </parent>

一、 pom.xml文件，引入依赖

 

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.springframework.boot</groupId>
  4.         <artifactId>spring-boot-starter-actuator</artifactId>
  5.     </dependency>
  6.     <dependency>
  7.         <groupId>org.springframework.boot</groupId>
  8.         <artifactId>spring-boot-starter-web</artifactId>
  9.     </dependency>
  10.     <dependency>
  11.         <groupId>org.springframework.kafka</groupId>
  12.         <artifactId>spring-kafka</artifactId>
  13.     </dependency>
  14.  
  15.     <dependency>
  16.         <groupId>org.springframework.boot</groupId>
  17.         <artifactId>spring-boot-starter-test</artifactId>
  18.         <scope>test</scope>
  19.     </dependency>
  20.     <dependency>
  21.         <groupId>org.springframework.kafka</groupId>
  22.         <artifactId>spring-kafka-test</artifactId>
  23.         <scope>test</scope>
  24.     </dependency>
  25. </dependencies>

采用Kafka提供的StringSerializer和StringDeserializer进行序列化和反序列化

1、在application-dev.properties配置生产者

 

  1. #============== kafka ===================
  2. # 指定kafka server的地址，集群配多个，中间，逗号隔开
  3. spring.kafka.bootstrap-servers=127.0.0.1:9092
  4.  
  5. #=============== provider  =======================
  6. # 写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败，
  7. # 当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。
  8. spring.kafka.producer.retries=0
  9. # 每次批量发送消息的数量,produce积累到一定数据，一次发送
  10. spring.kafka.producer.batch-size=16384
  11. # produce积累数据一次发送，缓存大小达到buffer.memory就发送数据
  12. spring.kafka.producer.buffer-memory=33554432
  13.  
  14. #procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
  15. #acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
  16. #acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
  17. #acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
  18. #可以设置的值为：all, -1, 0, 1
  19. spring.kafka.producer.acks=1
  20.  
  21. # 指定消息key和消息体的编解码方式
  22. spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  23. spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

  • bootstrap.servers:kafka server的地址

  • acks:写入kafka时，leader负责一个该partion读写，当写入partition时，需要将记录同步到repli节点，all是全部同步节点都返回成功，leader才返回ack。

  • retris:写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败，当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。

  • batch.size:produce积累到一定数据，一次发送。

buffer.memory: produce积累数据一次发送，缓存大小达到buffer.memory就发送数据。

  • linger.ms :当设置了缓冲区，消息就不会即时发送，如果消息总不够条数、或者消息不够buffer大小就不发送了吗？当消息超过linger时间，也会发送。

  • key/value serializer：序列化类。

2、生产者向kafka发送消息

 

  1. @RestController
  2. public class KafkaController {
  3.  
  4.     @Autowired
  5.     private KafkaTemplate<String,Object> kafkaTemplate;
  6.  
  7.     @GetMapping("/message/send")
  8.     public boolean send(@RequestParam String message){
  9.         kafkaTemplate.send("testTopic",message);
  10.         return true;
  11.     }
  12.  
  13. }

3、在application-dev.properties配置消费者

 

  1. #=============== consumer  =======================
  2. # 指定默认消费者group id --> 由于在kafka中，同一组中的consumer不会读取到同一个消息，依靠groud.id设置组名
  3. spring.kafka.consumer.group-id=testGroup
  4. # smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
  5. spring.kafka.consumer.auto-offset-reset=earliest
  6. # enable.auto.commit:true --> 设置自动提交offset
  7. spring.kafka.consumer.enable-auto-commit=true
  8. #如果'enable.auto.commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。
  9. spring.kafka.consumer.auto-commit-interval=100
  10.  
  11. # 指定消息key和消息体的编解码方式
  12. spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  13. spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

  • Producer是一个接口，声明了同步send和异步send两个重要方法。

  • ProducerRecord 消息实体类，每条消息由（topic,key,value,timestamp)四元组封装。一条消息key可以为空和timestamp可以设置当前时间为默认值。

4、消费者监听topic=testTopic的消息

 

  1. @Component
  2. public class ConsumerListener {
  3.  
  4.     @KafkaListener(topics = "testTopic")
  5.     public void onMessage(String message){
  6.         //insertIntoDb(buffer);//这里为插入数据库代码
  7.         System.out.println(message);
  8.     }
  9.  
  10. }

到此，采用Kafka提供的StringSerializer和StringDeserializer进行序列化和反序列化，因为此种序列化方式无法序列化实体类，顾，下面为自定义序列化和反序列化器进行实体类的消息传递

采用自定义序列化和反序列化器进行实体类的序列化和反序列化

和内置的StringSerializer字符串序列化一样，如果要自定义序列化方式，需要实现接口Serializer。假设每个字段按照下图所示的方式自定义序列化：

image

1、创建User实体类

 

  1. public class User implements Serializable {
  2.  
  3.     private Long id;
  4.  
  5.     private String name;
  6.  
  7.     private Integer age;
  8.  
  9.     /**
  10.      * transient 关键字修饰的字段不会被序列化
  11.      */
  12.     private transient String desc;
  13.  
  14.     public Long getId() {
  15.         return id;
  16.     }
  17.  
  18.     public void setId(Long id) {
  19.         this.id = id;
  20.     }
  21.  
  22.     public String getName() {
  23.         return name;
  24.     }
  25.  
  26.     public void setName(String name) {
  27.         this.name = name;
  28.     }
  29.  
  30.     public Integer getAge() {
  31.         return age;
  32.     }
  33.  
  34.     public void setAge(Integer age) {
  35.         this.age = age;
  36.     }
  37.  
  38.     public String getDesc() {
  39.         return desc;
  40.     }
  41.  
  42.     public void setDesc(String desc) {
  43.         this.desc = desc;
  44.     }
  45.  
  46.     @Override
  47.     public String toString() {
  48.         return "User{" +
  49.                 "id=" + id +
  50.                 ", name='" + name + '\'' +
  51.                 ", age=" + age +
  52.                 ", desc='" + desc + '\'' +
  53.                 '}';
  54.     }
  55. }

2、创建User序列化器

 

  1. public class UserSerializable implements Serializer<User> {
  2.     @Override
  3.     public void configure(Map<String, ?> map, boolean b) {
  4.  
  5.     }
  6.  
  7.     @Override
  8.     public byte[] serialize(String topic, User user) {
  9.         System.out.println("topic : " + topic + ", user : " + user);
  10.         byte[] dataArray = null;
  11.         ByteArrayOutputStream outputStream = null;
  12.         ObjectOutputStream objectOutputStream = null;
  13.         try {
  14.             outputStream = new ByteArrayOutputStream();
  15.             objectOutputStream = new ObjectOutputStream(outputStream);
  16.             objectOutputStream.writeObject(user);
  17.             dataArray = outputStream.toByteArray();
  18.         } catch (Exception e) {
  19.             throw new RuntimeException(e);
  20.         }finally {
  21.             if(outputStream != null){
  22.                 try {
  23.                     outputStream.close();
  24.                 } catch (IOException e) {
  25.                     e.printStackTrace();
  26.                 }
  27.             }
  28.             if(objectOutputStream != null){
  29.                 try {
  30.                     objectOutputStream.close();
  31.                 } catch (IOException e) {
  32.                     e.printStackTrace();
  33.                 }
  34.             }
  35.         }
  36.         return dataArray;
  37.     }
  38.  
  39.     @Override
  40.     public void close() {
  41.  
  42.     }
  43. }

3、创建User反序列化器

 

  1. public class UserDeserializer implements Deserializer<User> {
  2.     @Override
  3.     public void configure(Map<String, ?> map, boolean b) {
  4.  
  5.     }
  6.  
  7.     @Override
  8.     public User deserialize(String topic, byte[] bytes) {
  9.         User user = null;
  10.         ByteArrayInputStream inputStream = null;
  11.         ObjectInputStream objectInputStream = null;
  12.         try {
  13.             inputStream = new ByteArrayInputStream(bytes);
  14.             objectInputStream = new ObjectInputStream(inputStream);
  15.             user = (User)objectInputStream.readObject();
  16.         } catch (Exception e) {
  17.             throw new RuntimeException(e);
  18.         }finally {
  19.             if(inputStream != null){
  20.                 try {
  21.                     inputStream.close();
  22.                 } catch (IOException e) {
  23.                     e.printStackTrace();
  24.                 }
  25.             }
  26.             if(objectInputStream != null){
  27.                 try {
  28.                     objectInputStream.close();
  29.                 } catch (IOException e) {
  30.                     e.printStackTrace();
  31.                 }
  32.             }
  33.         }
  34.         return user;
  35.     }
  36.  
  37.     @Override
  38.     public void close() {
  39.  
  40.     }
  41. }

4、修改application-dev.properties配置

A、修改生产者配置的value-serializer

 

  1. # 指定生产者消息key和消息体的编解码方式
  2. spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  3. spring.kafka.producer.value-serializer=com.yibo.springbootkafkademo.Serializable.UserSerializable

B、修改消费者配置的value-deserializer

 

  1. # 指定消费者消息key和消息体的编解码方式
  2. spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  3. spring.kafka.consumer.value-deserializer=com.yibo.springbootkafkademo.Serializable.UserDeserializer

5、生产者向kafka发送消息

 

  1. @RestController
  2. public class KafkaController {
  3.  
  4.     @Autowired
  5.     private KafkaTemplate<String,Object> kafkaTemplate;
  6.  
  7.     @PostMapping("/user/save")
  8.     public boolean saveUser(@RequestBody User user){
  9.         kafkaTemplate.send("userTopic",user);
  10.         return true;
  11.     }
  12. }

6、消费者监听topic=userTopic的消息

 

  1. @Component
  2. public class ConsumerListener {
  3.  
  4.     @KafkaListener(topics = "userTopic")
  5.     public void onMessage(User user){
  6.         //insertIntoDb(buffer);//这里为插入数据库代码
  7.         System.out.println(user);
  8.     }
  9. }

总结

可以看到，自定义Serializer和Deserializer非常痛苦，还有很多类型不支持，非常脆弱。复杂类型的支持更是一件痛苦的事情，不同版本之间的兼容性问题更是一个极大的挑战。由于Serializer和Deserializer影响到上下游系统，导致牵一发而动全身。自定义序列化&反序列化实现不是能力的体现，而是逗比的体现。所以强烈不建议自定义实现序列化&反序列化，推荐直接使用StringSerializer和StringDeserializer，然后使用json作为标准的数据传输格式。站在巨人的肩膀上，事半功倍。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Guff_9hys/article/detail/870312
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号