Spring Boot Kafka - 序列化和反序列化JSON

亦凉 2022-09-05 06:11 662阅读 0赞

### 文章目录 ###

*  Spring Boot Kafka - 序列化和反序列化JSON
 *   *  前言
     *  配置JsonSerializer和JsonDeserializer
     *  定义一个Model类
     *  Producer类
     *  Consumer类
     *  Controller类
     *  测试
     *  小结
     *  参考文档

# Spring Boot Kafka - 序列化和反序列化JSON #

## 前言 ##

在[在Spring Boot应用集成Kafka读写消息][Spring Boot_Kafka] 一文中说明了如何通过Spring Kafka来发送和接收字符串消息。

本文描述了如何通过Spring Kafka如何在发送消息时序列化JSON（将对象转换成JSON字节流），如何在接收数据时反序列化JSON（将JSON字节流转换成对象）。

## 配置JsonSerializer和JsonDeserializer ##

在`application.yaml` 中配置JsonSerializer和JsonDeserializer。

server:
      port: 8080
    spring:
      kafka:
        consumer:
          bootstrap-servers: localhost:9092
          group-id: demo-group
          auto-offset-reset: earliest
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
          properties:
            spring:
              json:
                trusted:
                  packages: '*'
    
        producer:
          bootstrap-servers: localhost:9092
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

说明：

*  Consumer的`value-deserializer` 配置为 `org.springframework.kafka.support.serializer.JsonDeserializer`，表示在接收消息时用JSON反序列化。
 *  配置`spring.kafka.consumer.properties.spring.json.trusted.packages='*'` 表示接受反序列化任意的类，也可限定包路径。
 *  Producer的`value-serializer` 配置为`org.springframework.kafka.support.serializer.JsonSerializer`，表示在发送消息时用JSON序列化。

## 定义一个Model类 ##

定义一个`User`类。

@Data
    @NoArgsConstructor
    @AllArgsConstructor
    public class User { 
        private String firstName;
        private String lastName;
        private int age;
    }

说明：

*  通过Lombok注解来简化POJO的样板代码。

## Producer类 ##

定义一个`UserProducer`类。

@Service
    @RequiredArgsConstructor
    @Slf4j
    public class UserProducer { 
        private final KafkaTemplate<String, User> kafkaTemplate;
    
        public void createUser(User user) { 
            log.info(String.format("#### -> Producing message -> %s", user));
            kafkaTemplate.send(KafkaConstants.TOPIC_USERS, user);
        }
    }

说明：

*  `KafkaTemplate<String, User>` 表示message key为String，message value为User对象 （对象被序列化后为JSON字节流）。

## Consumer类 ##

定义一个`UserConsumer`类。

@Service
    @Slf4j
    public class UserConsumer { 
    
        @KafkaListener(topics = KafkaConstants.TOPIC_USERS, groupId = KafkaConstants.CONSUMER_GROUP)
        public void consume(User user) { 
            log.info(String.format("#### -> Consumed message -> %s", user));
        }
    }

说明：

*  `consume()` 方法的参数为User对象（JSON字节流被反序列化为对象）。

## Controller类 ##

定义一个`UserController`类用来作API测试。

@RestController
    @RequestMapping("/users")
    @RequiredArgsConstructor
    public class UserController { 
    
        private final UserProducer userProducer;
    
        @PostMapping
        public void createUser(@RequestBody User user) { 
            userProducer.createUser(user);
        }
    }

## 测试 ##

用HttPie工具来测试：

http :8080/users firstName="Cat" lastName="Tommy" age=3
    http :8080/users firstName="Mouse" lastName="Jackie" age=2

此时如果再测试之前的例子[在Spring Boot应用集成Kafka读写消息][Spring Boot_Kafka] ，也一样成功。

## 小结 ##

本文代码示例：

*  [https://github.com/cookcodeblog/spring-kafka-demo][https_github.com_cookcodeblog_spring-kafka-demo]

## 参考文档 ##

*  [Spring Kafka JOSN serialization and deserialization][]
 *  [Spring Boot Kafka JsonSerializer Example][]
 *  [Spring Boot Apache Kafka - JsonSerializer example][]

[Spring Boot_Kafka]: https://cookcode.blog.csdn.net/article/details/119730222
[https_github.com_cookcodeblog_spring-kafka-demo]: https://github.com/cookcodeblog/spring-kafka-demo
[Spring Kafka JOSN serialization and deserialization]: https://docs.spring.io/spring-kafka/reference/html/#json-serde
[Spring Boot Kafka JsonSerializer Example]: https://howtodoinjava.com/kafka/spring-boot-jsonserializer-example/
[Spring Boot Apache Kafka - JsonSerializer example]: https://www.linkedin.com/pulse/spring-boot-apache-kafka-jsonserializer-example-prateek-ashtikar/