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

spring:
 kafka:
 bootstrap-servers: 12.168.3.62:9092 # 指定kafka 代理地址，可以多個
 producer:
  retries: 2 # 寫入失敗時，重試次數(shù)。當retris為0時，produce不會重復。
  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在考慮完成請求之前收到的確認數(shù)，用于控制發(fā)送記錄在服務端的持久化，如果設置為零，則生產者將不會等待來自服務器的任何確認。
  key-serializer: org.apache.kafka.common.serialization.StringSerializer #指定消息key和消息體的編解碼方式
  value-serializer: org.apache.kafka.common.serialization.StringSerializer

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; //時間戳
}

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";
 }
}