一、添加依赖和添加配置项

1.1、在 Pom 文件中添加依赖

  org.springframework.kafka
  spring-kafka
 

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
 */
@Data
public 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
 */
@RestController
public class KafkaController {
 @Autowired
 private KafkaTemplate 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";
 }
}