# Spring Boot Kafka `JsonSerializer`示例 > 原文： [https://howtodoinjava.com/kafka/spring-boot-jsonserializer-example/](https://howtodoinjava.com/kafka/spring-boot-jsonserializer-example/) 學習使用[**`JsonSerializer`**](https://docs.spring.io/spring-kafka/api/org/springframework/kafka/support/serializer/JsonSerializer.html)和[`JsonDeserializer`](https://docs.spring.io/spring-kafka/api/org/springframework/kafka/support/serializer/JsonDeserializer.html)類從 [Apache Kafka](https://howtodoinjava.com/kafka/tutorial-introduction/) 主題存儲和檢索 JSON，并返回 Java 模型對象。 ## 1\. 先決條件 * 請按照本指南在您的機器上設置 [Kafka](https://howtodoinjava.com/kafka/getting-started-windows-10/) 。 * 我們正在修改 [Spring boot 和 Kafka HelloWorld 應用](https://howtodoinjava.com/kafka/spring-boot-with-kafka/)。 * 還要確保您的計算機上至少安裝了 [Java8](https://howtodoinjava.com/java/basics/jdk-jre-jvm/#downloads) 和 [Maven](https://howtodoinjava.com/maven/how-to-install-maven-on-windows/) 。 ## 2\. 應用配置 在`application.properties`文件中，我們添加了以下配置。 ```java server.port=9000 spring.kafka.consumer.bootstrap-servers: localhost:9092 spring.kafka.consumer.group-id: group-id spring.kafka.consumer.auto-offset-reset: earliest spring.kafka.consumer.key-deserializer: org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer spring.kafka.consumer.properties.spring.json.trusted.packages=* spring.kafka.producer.bootstrap-servers: localhost:9092 spring.kafka.producer.key-serializer: org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer: org.springframework.kafka.support.serializer.JsonSerializer ``` * `spring.kafka.consumer.key-deserializer`指定鍵的反序列化器類。 * `spring.kafka.consumer.value-deserializer`指定值的反序列化器類。 * `spring.kafka.consumer.properties.spring.json.trusted.packages`指定允許反序列化的程序包模式的逗號分隔列表。 `'*'`表示反序列化所有程序包。 * `spring.kafka.producer.key-deserializer`指定鍵的序列化器類。 * `spring.kafka.producer.value-deserializer`指定值的序列化器類。 ## 3\. 模型類 我們創建了`User`類，該類將發送給 Kafka。 其實例將由`JsonSerializer`序列化為字節數組。 這個字節數組最終由 Kafka 存儲到給定的分區中。 反序列化期間，`JsonDeserializer`用于以字節數組的形式從 Kafka 接收 JSON，并將`User`對象返回給應用。 ```java public class User { private long userId; private String firstName; private String lastName; public long getUserId() { return userId; } public void setUserId(long userId) { this.userId = userId; } public String getFirstName() { return firstName; } public void setFirstName(String firstName) { this.firstName = firstName; } public String getLastName() { return lastName; } public void setLastName(String lastName) { this.lastName = lastName; } @Override public String toString() { return "User [userId=" + userId + ", firstName=" + firstName + ", lastName=" + lastName + "]"; } } ``` ## 4\. Kafka 生產者 生產者 API 只是在 HTTP POST API 中使用用戶信息。 然后，它創建一個新的`User`對象，并使用[`KafkaTemplate`](https://docs.spring.io/spring-kafka/api/org/springframework/kafka/core/KafkaTemplate.html)發送給 Kafka。 ```java @PostMapping(value = "/createUser") public void sendMessageToKafkaTopic( @RequestParam("userId") long userId, @RequestParam("firstName") String firstName, @RequestParam("lastName") String lastName) { User user = new User(); user.setUserId(userId); user.setFirstName(firstName); user.setLastName(lastName); this.producerService.saveCreateUserLog(user); } ``` ```java @Autowired private KafkaTemplate<String, Object> kafkaTemplate; public void saveCreateUserLog(User user) { logger.info(String.format("User created -> %s", user)); this.kafkaTemplate.send(AppConstants.TOPIC_NAME_USER_LOG, user); } ``` ## 5\. Kafka 消費者 用戶被實現為`@KafkaListener`，每次在主題中添加新條目時都會得到通知。 ```java @KafkaListener(topics = AppConstants.TOPIC_NAME_USER_LOG, groupId = AppConstants.GROUP_ID) public void consume(User user) { logger.info(String.format("User created -> %s", user)); } ``` ## 6\. 測試 使用任何 REST API 測試器，并向 API `http://localhost:9000/kafka/createUser`發送一些消息，如下所示。 留言欄：`http://localhost:9000/kafka/createUser?userId=1&firstName=Lokesh&lastName=Gupta` 觀察控制臺日志： ```java 2020-05-24 23:36:47.132 INFO 2092 --- [nio-9000-exec-4] 2020-05-26 01:03:52.722 INFO 11924 --- [nio-9000-exec-6] c.h.k.demo.service.KafKaProducerService : User created -> User [userId=1, firstName=Lokesh, lastName=Gupta] 2020-05-26 01:03:52.729 INFO 11924 --- [ntainer#1-0-C-1] c.h.k.demo.service.KafKaConsumerService : User created -> User [userId=1, firstName=Lokesh, lastName=Gupta] ``` ## 7\. 總結 在此 SpringBoot kafka `JsonSerializer`示例中，我們學習了使用`JsonSerializer`對 Java 對象進行序列化和反序列化并存儲在 Kafka 中。 學習愉快！ [源碼下載](https://github.com/lokeshgupta1981/Kafka-Tutorials)