快捷導(dǎo)航

Spring boot 整合KAFKA消息隊列的示例

更新時間：2020年10月16日 11:33:50 作者：拾階求上

這篇文章主要介紹了Spring boot 整合 KAFKA 消息隊列的示例，幫助大家更好的理解和使用spring boot框架，感興趣的朋友可以了解下

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

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

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 # 寫入失敗時，重試次數(shù)。當(dāng)retris為0時，produce不會重復(fù)。
  batch-size: 1000 #每次批量發(fā)送消息的數(shù)量,produce積累到一定數(shù)據(jù)，一次發(fā)送
  buffer-memory: 33554432 # produce積累數(shù)據(jù)一次發(fā)送，緩存大小達到buffer.memory就發(fā)送數(shù)據(jù)
  acks: 0 #procedure要求leader在考慮完成請求之前收到的確認(rèn)數(shù)，用于控制發(fā)送記錄在服務(wù)端的持久化，如果設(shè)置為零，則生產(chǎn)者將不會等待來自服務(wù)器的任何確認(rèn)。
  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
 */
@Data
public class Message {
 private Long id; //id

 private String msg; //消息

 private Date sendTime; //時間戳
}

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
 */
@RestController
public 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 消息隊列的示例的詳細內(nèi)容，更多關(guān)于Spring boot 整合消息隊列的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: