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

server.port=80
#kafka地址，可以有多個(gè)
spring.kafka.bootstrap-servers=127.0.0.1:9092
#------生產(chǎn)者配置文件---------
#指定kafka消息體和key的編碼格式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
#設(shè)置等待acks返回的機(jī)制，有三個(gè)值
# 0：不等待返回的acks（可能會(huì)丟數(shù)據(jù)，因?yàn)榘l(fā)送消息沒(méi)有了失敗重試機(jī)制，但是這是最低延遲）
# 1：消息發(fā)送給kafka分區(qū)中的leader后就返回（如果follower沒(méi)有同步完成leader就宕機(jī)了，就會(huì)丟數(shù)據(jù)）
# -1（默認(rèn)）：等待所有follower同步完消息后再發(fā)送（絕對(duì)不會(huì)丟數(shù)據(jù)）
spring.kafka.producer.acks=-1
# 消息累計(jì)到batch-size的值后，才會(huì)發(fā)送消息，默認(rèn)為16384
spring.kafka.producer.batch-size=16384
#如果kafka遲遲不發(fā)送消息（這里指的是消息沒(méi)堆積到指定數(shù)量），那么過(guò)了這個(gè)時(shí)間（單位：毫米）開(kāi)始發(fā)送
spring.kafka.producer.properties.linger.ms=1
#設(shè)置緩沖區(qū)大小，默認(rèn)是33554432
#這個(gè)緩沖區(qū)是kafka中兩個(gè)線程里的共享變量
#這個(gè)兩個(gè)線程是main和sender，main負(fù)責(zé)把消息發(fā)送到共享變量，sender從共享變量拉數(shù)據(jù)
spring.kafka.producer.buffer-memory=33554432
#失敗重試發(fā)送的次數(shù)
spring.kafka.producer.retries=2
#------消費(fèi)者配置文件---------
#指定kafka消息體和key的編碼格式
spring.kafka.consumer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.consumer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
#指定消費(fèi)者組的group_id
spring.kafka.consumer.group-id=kafka_test
#kafka意外宕機(jī)時(shí)，再次開(kāi)啟消息消費(fèi)的策略，共有三種策略
#earliest：當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開(kāi)始消費(fèi)；無(wú)提交的offset時(shí)，從頭開(kāi)始消費(fèi)
#latest：當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開(kāi)始消費(fèi)；無(wú)提交的offset時(shí)，消費(fèi)新產(chǎn)生的該分區(qū)下的數(shù)據(jù)、
#none：當(dāng)所有分區(qū)都存在已提交的offset時(shí)，從offset后開(kāi)始消費(fèi)；只要有一個(gè)分區(qū)不存在已提交的offset，則拋出異常
spring.kafka.consumer.auto-offset-reset=earliest
#自動(dòng)提交offset
spring.kafka.consumer.enable-auto-commit=true
#提交offset時(shí)間間隔
spring.kafka.consumer.auto-commit-interval=100
#消費(fèi)監(jiān)聽(tīng)接口監(jiān)聽(tīng)的主題不存在時(shí)，默認(rèn)會(huì)報(bào)錯(cuò)因此要關(guān)掉這個(gè)
spring.kafka.listener.missing-topics-fatal=false

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 使用代碼創(chuàng)建的topic
 * 三個(gè)參數(shù)意思：topic的名稱；分區(qū)數(shù)量，新主題的復(fù)制因子；如果指定了副本分配，則為-1。
 */
@Configuration
public class KafkaTopic {

   @Bean
  public NewTopic batchTopic() {
    return new NewTopic("testTopic", 8, (short) 1);
  }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * kafka生產(chǎn)者代碼
 */
@RestController
public class ProductorController {

  @Autowired
  private KafkaTemplate<String, String> kafkaTemplate;

  @RequestMapping("/test")
  public String show() {
    kafkaTemplate.send("testTopic", "你好");
    return "發(fā)送成功";
  }

}

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;

import java.util.Optional;

/**
 * kafka消費(fèi)者代碼
 */
@Configuration
public class KafkaConsumer {
  
  @KafkaListener(topics = "testTopic")
  public void consumer(ConsumerRecord consumerRecord){
    Optional<Object> kafkaMassage = Optional.ofNullable(consumerRecord.value());
    if(kafkaMassage.isPresent()){
      Object o = kafkaMassage.get();
      System.out.println("接收到的消息是："+o);
    }

  }

}