這里使用 spring-kafka 依賴和 KafkaTemplate 對象來操作 Kafka 服務。

一、添加依賴和添加配置項

1.1、在 Pom 文件中添加依賴

<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>

1.2、添加配置項

spring: kafka: bootstrap-servers: 12.168.3.62:9092 # 指定kafka 代理地址，可以多個 producer: retries: 2 # 寫入失敗時，重試次數。當retris為0時，produce不會重復。 batch-size: 1000 #每次批量發送消息的數量,produce積累到一定數據，一次發送 buffer-memory: 33554432 # produce積累數據一次發送，緩存大小達到buffer.memory就發送數據 acks: 0 #procedure要求leader在考慮完成請求之前收到的確認數，用于控制發送記錄在服務端的持久化，如果設置為零，則生產者將不會等待來自服務器的任何確認。 key-serializer: org.apache.kafka.common.serialization.StringSerializer #指定消息key和消息體的編解碼方式 value-serializer: org.apache.kafka.common.serialization.StringSerializer

二、代碼編寫

2.1、添加一個消息類

package com.jsh.mgt.kafkaTemplate.kafka;import java.util.Date;import lombok.Data;/** * @since 2020/5/21 14:13 */@Datapublic class Message { private Long id; //id private String msg; //消息 private Date sendTime; //時間戳}

2.2、設置消息生產者

package com.jsh.mgt.kafkaTemplate.Controllers;import com.google.gson.Gson;import com.google.gson.GsonBuilder;import com.jsh.mgt.kafkaTemplate.kafka.Message;import java.util.Date;import java.util.UUID;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.kafka.core.KafkaTemplate;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.PathVariable;import org.springframework.web.bind.annotation.RestController;/** * @since 2020/5/21 11:19 */@RestControllerpublic class KafkaController { @Autowired private KafkaTemplate<String,Object> kafkaTemplate; private Gson gson = new GsonBuilder().create(); @GetMapping('/kafka/{msg}') public Object test(@PathVariable('msg') String msg) { Message message = new Message(); message.setId(System.currentTimeMillis()); message.setMsg(UUID.randomUUID().toString()+ '-'+msg); message.setSendTime(new Date()); kafkaTemplate.send('topic-create',gson.toJson(message)); return 'ok'; }}

以上就是Spring boot 整合 KAFKA 消息隊列的示例的詳細內容，更多關于Spring boot 整合消息隊列的資料請關注好吧啦網其它相關文章！