.\bin\windows\zookeeper-server-start.bat config\zookeeper.properties

.\bin\windows\kafka-server-start.bat config\server.properties

# 引入jar包
<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>

spring.kafka.producer.bootstrap-servers=localhost:9092 # 配置kafka地址
spring.kafka.consumer.max-poll-records=1 #每次拉取一条数据 方便测试
spring.kafka.consumer.heartbeat-interval=100 # 心跳时间100ms

package com.example.kafka_test.service.impl;
import com.example.kafka_test.service.ProducerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import static com.example.kafka_test.KafkaConstant.TOPIC_NAME;
/**
 * @Author: Zy
 * @Date: 2021/11/25 10:00
 */
@Service
public class ProducerServiceImpl implements ProducerService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Override
    public void sendMsg(String msg) {
        kafkaTemplate.send(TOPIC_NAME, msg);
    }
}

package com.example.kafka_test.service.impl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import static com.example.kafka_test.KafkaConstant.CONSUMER_GROUP;
import static com.example.kafka_test.KafkaConstant.TOPIC_NAME;
/**
 * @Author: Zy
 * @Date: 2021/11/25 10:13
 */
@Slf4j
@Component
public class ConsumerServiceImpl {
    @KafkaListener(topics = TOPIC_NAME, groupId = CONSUMER_GROUP)
    public void consumerMsg(String msg) {
        try {
            # 休眠 防止消费速度快 无法观察日志
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info(msg);
    }
}