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

spring:
  # Kafka 配置
  kafka:
    # Kafka 服務(wù)器地址和端口 代理地址，可以多個
    bootstrap-servers: IP:9092
    # 生產(chǎn)者配置
    producer:
      # 發(fā)送失敗時的重試次數(shù)
      retries: 3
      # 每次批量發(fā)送消息的數(shù)量，調(diào)整為較小值
      batch-size: 1
      # 生產(chǎn)者緩沖區(qū)大小
      buffer-memory: 33554432
      # 消息 key 的序列化器，將 key 序列化為字節(jié)數(shù)組
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 消息 value 的序列化器，將消息體序列化為字節(jié)數(shù)組
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    # 消費(fèi)者配置
    consumer:
      # 當(dāng)沒有初始偏移量或當(dāng)前偏移量不存在時，從最早的消息開始消費(fèi)
      auto-offset-reset: earliest
      # 是否自動提交偏移量
      enable-auto-commit: true
      # 自動提交偏移量的時間間隔（毫秒），延長自動提交時間間隔
      auto-commit-interval: 1000
      # 消息 key 的反序列化器，將字節(jié)數(shù)組反序列化為 key
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 消息 value 的反序列化器，將字節(jié)數(shù)組反序列化為消息體
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
/**
 * 生產(chǎn)者
 *
 * @author chenlei
 */
@Slf4j
@Component
public class KafkaProducer {
    /**
     * KafkaTemplate
     */
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    /**
     * 發(fā)送消息到指定的 Kafka 主題，并可指定分組信息
     *
     * @param topic   消息要發(fā)送到的 Kafka 主題
     * @param message 要發(fā)送的消息內(nèi)容
     */
    public void sendMessage(String topic, String message) {
        // 使用 KafkaTemplate 發(fā)送消息，將消息發(fā)送到指定的主題
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                // 消息發(fā)送成功后的處理邏輯，可根據(jù)需要添加
                log.info("已發(fā)送消息=[" + message + "]，其偏移量=[" + result.getRecordMetadata().offset() + "]");
            }
            @Override
            public void onFailure(Throwable ex) {
                // 消息發(fā)送失敗后的處理邏輯，使用日志記錄異常
                log.error("發(fā)送消息=[" + message + "] 失敗", ex);
            }
        });
    }
}

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
/**
 * @author 消費(fèi)者
 * chenlei
 */
@Slf4j
@Component
public class KafkaConsumer {
    /**
     * 監(jiān)聽 Kafka 主題方法。
     *
     * @param record 從 Kafka 接收到的 ConsumerRecord，包含消息的鍵值對
     */
    @KafkaListener(topics = {"topic"}, groupId = "consumer.group-id", concurrency = "5")
    public void listen(ConsumerRecord<?, ?> record) {
        // 打印接收到的消息的詳細(xì)信息
        log.info("接收到 Kafka 消息: 主題 = {}, 分區(qū) = {}, 偏移量 = {}, 鍵 = {}, 值 = {}",
                record.topic(), record.partition(), record.offset(), record.key(), record.value());
    }
}