這里使用 spring-kafka 依賴和 KafkaTemplate 對(duì)象來(lái)操作 Kafka 服務(wù)。

一、添加依賴和添加配置項(xiàng)

1.1、在 Pom 文件中添加依賴

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

1.2、添加配置項(xiàng)

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

二、代碼編寫(xiě)

2.1、添加一個(gè)消息類(lèi)

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; //時(shí)間戳}

2.2、設(shè)置消息生產(chǎn)者

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 消息隊(duì)列的示例的詳細(xì)內(nèi)容，更多關(guān)于Spring boot 整合消息隊(duì)列的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！