在Spring Boot應用程序中使用Apache Kafka的方法步驟詳解

2024-07-14 08:42:45

字體：大中小

來源：轉載

供稿：網友

第1步：生成我們的項目： Spring Initializr 來生成我們的項目。我們的項目將提供Spring MVC / Web支持和Apache Kafka支持。

第2步：發布/讀取Kafka主題中的消息：

<b>public</b> <b>class</b> User {  <b>private</b> String name;  <b>private</b> <b>int</b> age;  <b>public</b> User(String name, <b>int</b> age) {    <b>this</b>.name = name;    <b>this</b>.age = age;  }}

第3步：通過application.yml配置文件配置Kafka:

我們需要創建配置文件。我們需要以某種方式配置我們的Kafka生產者和消費者，以便能夠發布和讀取與主題相關的消息。相比建立一個使用@Configuration標注的Java類，我們可以直接使用配置文件application.properties或application.yml。Spring Boot讓我們避免像過去一樣編寫的所有樣板代碼，同時為我們提供了更加智能的配置應用程序的方法，如下所示：

server: port: 9000spring:  kafka:   consumer:    bootstrap: 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.StringDeserializerproducer:    bootstrap: localhost:9092    key-serializer: org.apache.kafka.common.serialization.StringSerializer    value-serializer: org.apache.kafka.common.serialization.StringSerializer

第4步：創建一個生產者，創建生產者會將我們的消息寫入該主題。

<b>public</b> <b>class</b> Producer {  <b>private</b> <b>static</b> <b>final</b> Logger logger = LoggerFactory.getLogger(Producer.<b>class</b>);  <b>private</b> <b>static</b> <b>final</b> String TOPIC = <font>"users"</font><font>;  @Autowired  <b>private</b> KafkaTemplate<String, String> kafkaTemplate;  <b>public</b> <b>void</b> sendMessage(String message) {    logger.info(String.format(</font><font>"#### -> Producing message -> %s"</font><font>, message));    <b>this</b>.kafkaTemplate.send(TOPIC, message);  }}</font>

自動連接autowire到 KafkaTemplate ，使用它將消息發布到主題 - 這就是消息的生產者！

第5步：創建一個消費者，消費者是負責根據您自己的業務邏輯的需求閱讀處理消息的消息的服務。要進行設置，請輸入以下內容：

@Service<b>public</b> <b>class</b> Consumer {  <b>private</b> <b>final</b> Logger logger = LoggerFactory.getLogger(Producer.<b>class</b>);  @KafkaListener(topics = <font>"users"</font><font>, groupId = </font><font>"group_id"</font><font>)  <b>public</b> <b>void</b> consume(String message) throws IOException {    logger.info(String.format(</font><font>"#### -> Consumed message -> %s"</font><font>, message));  }}</font>

在這里，我們告訴我們的方法void consume（String message）訂閱用戶的主題，并將每條消息發送到應用程序日志。在您的實際應用程序中，您可以按照業務需要的方式處理消息。

第6步：創建REST控制器，們已經擁有了能夠消費Kafka消息所需的全部內容。

為了充分展示我們創建的所有內容的工作原理，我們需要創建一個具有單一端點的控制器。消息將發布到此端點，然后由我們的生產者處理。然后，我們的消費者將通過登錄到控制臺來捕獲并處理它。

@RestController@RequestMapping(value = <font>"/kafka"</font><font>)<b>public</b> <b>class</b> KafkaController {  <b>private</b> <b>final</b> Producer producer;  @Autowired  KafkaController(Producer producer) {    <b>this</b>.producer = producer;  }  @PostMapping(value = </font><font>"/publish"</font><font>)  <b>public</b> <b>void</b> sendMessageToKafkaTopic(@RequestParam(</font><font>"message"</font><font>) String message) {    <b>this</b>.producer.sendMessage(message);  }}</font>

讓我們使用cURL將消息發送給Kafka：