# Kafka 的 Spring Boot – HelloWorld 示例 > 原文： [https://howtodoinjava.com/kafka/spring-boot-with-kafka/](https://howtodoinjava.com/kafka/spring-boot-with-kafka/) 學習創建一個 [SpringBoot](https://howtodoinjava.com/spring-boot-tutorials/) 應用，該應用能夠連接給定的 [Apache Kafka](https://howtodoinjava.com/kafka/tutorial-introduction/) Borker 實例。 另外，從 Kafka 主題中學習產生和消費消息。 我們將遵循的步驟： * 創建具有 Kafka 依賴項的 Spring Boot 應用 * 在`application.yaml`中配置 kafka Borker 實例 * 使用[`KafkaTemplate`](https://docs.spring.io/spring-kafka/api/org/springframework/kafka/core/KafkaTemplate.html)將消息發送到主題 * 使用[`@KafkaListener`](https://docs.spring.io/spring-kafka/api/org/springframework/kafka/annotation/KafkaListener.html)實時收聽發送到主題的消息 ## 1\. 先決條件 * 請按照本指南在您的機器上設置 [Kafka](https://howtodoinjava.com/kafka/getting-started-windows-10/) 。 * 我們正在創建一個基于 Maven 的 Spring 引導應用，因此您的計算機應至少安裝 [Java8](https://howtodoinjava.com/java/basics/jdk-jre-jvm/#downloads) 和 [Maven](https://howtodoinjava.com/maven/how-to-install-maven-on-windows/) 。 ## 2\. Spring Boot 應用 打開 [spring 初始化器](https://start.spring.io/)并創建具有以下依賴項的 spring boot 應用： * Spring Apache Kafka * Spring Web  創建 Spring boot kafka 應用 生成的項目在`pom.xml`中具有以下依賴項。 ```java <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> <exclusions> <exclusion> <groupId>org.junit.vintage</groupId> <artifactId>junit-vintage-engine</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka-test</artifactId> <scope>test</scope> </dependency> </dependencies> ``` 在首選的 IDE 中導入項目。 ## 3\. 配置 Kafka Borker 在`application.yaml`文件中，添加 kafka Borker 地址以及與消費者和生產者相關的配置。 ```java server: port: 9000 spring: kafka: consumer: bootstrap-servers: localhost:9092 group-id: group-id auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer producer: bootstrap-servers: localhost:9092 key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer ``` ## 4\. `KafKaProducerService`和`KafKaConsumerService` `KafKaProducerService`類使用自動連接的`KafkaTemplate`將消息發送到已配置的主題名稱。 同樣，`KafKaConsumerService`類使用`@KafkaListener`來接收來自已配置主題名稱的消息。 ```java import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Service; import com.howtodoinjava.kafka.demo.common.AppConstants; @Service public class KafKaProducerService { private static final Logger logger = LoggerFactory.getLogger(KafKaProducerService.class); @Autowired private KafkaTemplate<String, String> kafkaTemplate; public void sendMessage(String message) { logger.info(String.format("Message sent -> %s", message)); this.kafkaTemplate.send(AppConstants.TOPIC_NAME, message); } } ``` ```java import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Service; import com.howtodoinjava.kafka.demo.common.AppConstants; @Service public class KafKaConsumerService { private final Logger logger = LoggerFactory.getLogger(KafKaConsumerService.class); @KafkaListener(topics = AppConstants.TOPIC_NAME, groupId = AppConstants.GROUP_ID) public void consume(String message) { logger.info(String.format("Message recieved -> %s", message)); } } ``` ```java public class AppConstants { public static final String TOPIC_NAME = "test"; public static final String GROUP_ID = "group_id"; } ``` ## 5\. 控制器 控制器負責使用 REST API 從用戶獲取消息，并將消息移交給生產者服務以將其發布到 kafka 主題。 ```java import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import com.howtodoinjava.kafka.demo.service.KafKaProducerService; @RestController @RequestMapping(value = "/kafka") public class KafkaProducerController { private final KafKaProducerService producerService; @Autowired public KafkaProducerController(KafKaProducerService producerService) { this.producerService = producerService; } @PostMapping(value = "/publish") public void sendMessageToKafkaTopic(@RequestParam("message") String message) { this.producerService.sendMessage(message); } } ``` ## 6\. 測試 使用任何 REST API 測試器并將少量消息發布到查詢參數`"message"`中的 API `http://localhost:9000/kafka/publish`。 留言欄：`http://localhost:9000/kafka/publish?message=Alphabet` 觀察控制臺日志： ```java 2020-05-24 23:36:47.132 INFO 2092 --- [nio-9000-exec-4] c.h.k.demo.service.KafKaProducerService : Message sent -> Alphabet 2020-05-24 23:36:47.138 INFO 2092 --- [ntainer#0-0-C-1] c.h.k.demo.service.KafKaConsumerService : Message recieved -> Alphabet ``` 如果您已經在命令提示符下打開了 Kafka 控制臺用戶，那么您也會看到該消息。  Kafka 控制臺消費者 ## 7\. 總結 在 **spring boot kafka 教程**中，我們學習了如何創建 spring boot 應用和配置 Kafka 服務器。 另外，我們通過使用`KafkaTemplate`發布一些消息，然后使用`@KafkaListener`使用這些消息來驗證該應用。 學習愉快！ [源碼下載](https://github.com/lokeshgupta1981/Kafka-Tutorials)