# 對于Ubuntu/Debian
sudo apt-get update
sudo apt-get install openjdk-17-jdk

# 對于CentOS/RHEL
sudo yum install java-17-openjdk

wget https://downloads.apache.org/kafka/3.6.1/kafka_2.13-3.6.1.tgz
tar -xzf kafka_2.13-3.6.1.tgz
cd kafka_2.13-3.6.1

# 后臺啟動ZooKeeper
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

# 后臺啟動Kafka
bin/kafka-server-start.sh -daemon config/server.properties

bin/kafka-topics.sh --create --topic test-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

docker-compose up -d

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.2.0</version>
        <relativePath/>
    </parent>
    
    <groupId>com.jam</groupId>
    <artifactId>springboot-kafka-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-kafka-demo</name>
    <description>SpringBoot集成Kafka示例項目</description>
    
    <properties>
        <java.version>17</java.version>
        <lombok.version>1.18.30</lombok.version>
        <commons-lang3.version>3.14.0</commons-lang3.version>
        <mybatis-plus.version>3.5.5</mybatis-plus.version>
        <mysql-connector.version>8.2.0</mysql-connector.version>
        <springdoc.version>2.1.0</springdoc.version>
        <kafka.version>3.6.1</kafka.version>
    </properties>
    
    <dependencies>
        <!-- SpringBoot核心依賴 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        
        <!-- Kafka依賴 -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>${kafka.version}</version>
        </dependency>
        
        <!-- Lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>${lombok.version}</version>
            <scope>provided</scope>
        </dependency>
        
        <!-- 工具類 -->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>${commons-lang3.version}</version>
        </dependency>
        
        <!-- MyBatis-Plus -->
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>${mybatis-plus.version}</version>
        </dependency>
        
        <!-- MySQL驅(qū)動 -->
        <dependency>
            <groupId>com.mysql</groupId>
            <artifactId>mysql-connector-j</artifactId>
            <version>${mysql-connector.version}</version>
            <scope>runtime</scope>
        </dependency>
        
        <!-- Swagger3 -->
        <dependency>
            <groupId>org.springdoc</groupId>
            <artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
            <version>${springdoc.version}</version>
        </dependency>
        
        <!-- 測試依賴 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <version>${kafka.version}</version>
            <scope>test</scope>
        </dependency>
    </dependencies>
    
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

spring:
  application:
    name: springboot-kafka-demo
  datasource:
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://localhost:3306/kafka_demo?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai
    username: root
    password: root
  kafka:
    # Kafka集群地址
    bootstrap-servers: localhost:9092
    # 生產(chǎn)者配置
    producer:
      # 消息key的序列化器
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 消息value的序列化器
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      # 批次大小，當(dāng)批次滿了之后才會發(fā)送
      batch-size: 16384
      # 緩沖區(qū)大小
      buffer-memory: 33554432
      # 消息確認(rèn)機(jī)制：0-不需要確認(rèn)，1-只需要leader確認(rèn)，all-所有副本都需要確認(rèn)
      acks: all
      # 重試次數(shù)
      retries: 3
      # 重試間隔時間
      retry-backoff-ms: 1000
    # 消費(fèi)者配置
    consumer:
      # 消息key的反序列化器
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 消息value的反序列化器
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      # 消費(fèi)者組ID
      group-id: default-group
      # 自動偏移量重置策略：earliest-從頭開始消費(fèi)，latest-從最新的開始消費(fèi)，none-如果沒有偏移量則拋出異常
      auto-offset-reset: earliest
      # 是否自動提交偏移量
      enable-auto-commit: false
      # 自動提交偏移量的間隔時間
      auto-commit-interval: 1000
      # 指定JsonDeserializer反序列化的目標(biāo)類
      properties:
        spring:
          json:
            trusted:
              packages: com.jam.entity
    # 監(jiān)聽器配置
    listener:
      # 消息確認(rèn)模式：manual-手動確認(rèn)，auto-自動確認(rèn)
      ack-mode: manual_immediate
      # 并發(fā)消費(fèi)者數(shù)量
      concurrency: 3
      # 批量消費(fèi)配置
      batch-listener: false
      # 每次拉取的記錄數(shù)
      consumer:
        max-poll-records: 500
      # 重試配置
      retry:
        # 是否啟用重試
        enabled: true
        # 初始重試間隔時間
        initial-interval: 1000
        # 最大重試間隔時間
        max-interval: 10000
        # 重試乘數(shù)
        multiplier: 2
        # 最大重試次數(shù)
        max-attempts: 3

mybatis-plus:
  mapper-locations: classpath:mapper/*.xml
  type-aliases-package: com.jam.entity
  configuration:
    map-underscore-to-camel-case: true
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

springdoc:
  api-docs:
    path: /api-docs
  swagger-ui:
    path: /swagger-ui.html
    operationsSorter: method

server:
  port: 8081

package com.jam.config;

/**
 * Kafka常量配置類
 * 定義Kafka主題名稱、消費(fèi)者組等常量
 *
 * @author 果醬
 */
public class KafkaConstant {
    /**
     * 普通消息主題
     */
    public static final String NORMAL_TOPIC = "normal_topic";
    
    /**
     * 分區(qū)消息主題
     */
    public static final String PARTITION_TOPIC = "partition_topic";
    
    /**
     * 事務(wù)消息主題
     */
    public static final String TRANSACTIONAL_TOPIC = "transactional_topic";
    
    /**
     * 死信主題
     */
    public static final String DEAD_LETTER_TOPIC = "dead_letter_topic";
    
    /**
     * 普通消費(fèi)者組
     */
    public static final String NORMAL_CONSUMER_GROUP = "normal_consumer_group";
    
    /**
     * 分區(qū)消費(fèi)者組
     */
    public static final String PARTITION_CONSUMER_GROUP = "partition_consumer_group";
    
    /**
     * 事務(wù)消費(fèi)者組
     */
    public static final String TRANSACTIONAL_CONSUMER_GROUP = "transactional_consumer_group";
    
    /**
     * 死信消費(fèi)者組
     */
    public static final String DEAD_LETTER_CONSUMER_GROUP = "dead_letter_consumer_group";
    
    /**
     * 事務(wù)ID前綴
     */
    public static final String TRANSACTION_ID_PREFIX = "tx-";
}

package com.jam.entity;

import lombok.Data;
import java.io.Serializable;
import java.time.LocalDateTime;

/**
 * 消息實(shí)體類
 * 用于封裝發(fā)送到Kafka的消息內(nèi)容
 *
 * @author 果醬
 */
@Data
public class MessageEntity implements Serializable {
    /**
     * 消息ID
     */
    private String messageId;
    
    /**
     * 消息內(nèi)容
     */
    private String content;
    
    /**
     * 業(yè)務(wù)類型
     */
    private String businessType;
    
    /**
     * 業(yè)務(wù)ID，用于分區(qū)策略
     */
    private String businessId;
    
    /**
     * 創(chuàng)建時間
     */
    private LocalDateTime createTime;
    
    /**
     * 擴(kuò)展字段，用于存儲額外信息
     */
    private String extra;
}

package com.jam.config;

import com.jam.entity.MessageEntity;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import java.util.HashMap;
import java.util.Map;
import static org.apache.kafka.clients.consumer.ConsumerConfig.*;
import static org.apache.kafka.clients.producer.ProducerConfig.*;

/**
 * Kafka配置類
 * 配置Kafka主題、生產(chǎn)者、消費(fèi)者等
 *
 * @author 果醬
 */
@Configuration
public class KafkaConfig {
    /**
     * 創(chuàng)建普通消息主題
     * 3個分區(qū)，1個副本
     *
     * @return 普通消息主題
     */
    @Bean
    public NewTopic normalTopic() {
        // 參數(shù)：主題名稱、分區(qū)數(shù)、副本數(shù)
        return new NewTopic(KafkaConstant.NORMAL_TOPIC, 3, (short) 1);
    }
    
    /**
     * 創(chuàng)建分區(qū)消息主題
     * 5個分區(qū)，1個副本
     *
     * @return 分區(qū)消息主題
     */
    @Bean
    public NewTopic partitionTopic() {
        return new NewTopic(KafkaConstant.PARTITION_TOPIC, 5, (short) 1);
    }
    
    /**
     * 創(chuàng)建事務(wù)消息主題
     * 3個分區(qū)，1個副本
     *
     * @return 事務(wù)消息主題
     */
    @Bean
    public NewTopic transactionalTopic() {
        return new NewTopic(KafkaConstant.TRANSACTIONAL_TOPIC, 3, (short) 1);
    }
    
    /**
     * 創(chuàng)建死信主題
     * 1個分區(qū)，1個副本
     *
     * @return 死信主題
     */
    @Bean
    public NewTopic deadLetterTopic() {
        return new NewTopic(KafkaConstant.DEAD_LETTER_TOPIC, 1, (short) 1);
    }
    
    /**
     * 配置事務(wù)生產(chǎn)者工廠
     *
     * @return 事務(wù)生產(chǎn)者工廠
     */
    @Bean
    public ProducerFactory<String, MessageEntity> transactionalProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
        configProps.put(VALUE_SERIALIZER_CLASS_CONFIG, org.springframework.kafka.support.serializer.JsonSerializer.class);
        configProps.put(ACKS_CONFIG, "all");
        configProps.put(RETRIES_CONFIG, 3);
        configProps.put(BATCH_SIZE_CONFIG, 16384);
        configProps.put(BUFFER_MEMORY_CONFIG, 33554432);
        // 配置事務(wù)ID前綴
        configProps.put(TRANSACTIONAL_ID_CONFIG, KafkaConstant.TRANSACTION_ID_PREFIX);
        
        DefaultKafkaProducerFactory<String, MessageEntity> factory = 
            new DefaultKafkaProducerFactory<>(configProps);
        // 開啟事務(wù)支持
        factory.transactionCapable();
        return factory;
    }
    
    /**
     * 配置事務(wù)Kafka模板
     *
     * @return 事務(wù)Kafka模板
     */
    @Bean
    public KafkaTemplate<String, MessageEntity> transactionalKafkaTemplate() {
        return new KafkaTemplate<>(transactionalProducerFactory());
    }
    
    /**
     * 配置Kafka事務(wù)管理器
     *
     * @return Kafka事務(wù)管理器
     */
    @Bean
    public KafkaTransactionManager<String, MessageEntity> kafkaTransactionManager() {
        return new KafkaTransactionManager<>(transactionalProducerFactory());
    }
}

package com.jam.config;

import com.jam.entity.MessageEntity;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.InvalidRecordException;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.utils.Utils;
import java.util.List;
import java.util.Map;

/**
 * 自定義Kafka分區(qū)策略
 * 根據(jù)業(yè)務(wù)ID將消息發(fā)送到指定分區(qū)，確保相同業(yè)務(wù)ID的消息在同一分區(qū)
 *
 * @author 果醬
 */
public class BusinessIdPartitioner implements Partitioner {
    /**
     * 計算分區(qū)號
     *
     * @param topic 主題名稱
     * @param key 消息鍵
     * @param keyBytes 消息鍵的字節(jié)數(shù)組
     * @param value 消息值
     * @param valueBytes 消息值的字節(jié)數(shù)組
     * @param cluster Kafka集群信息
     * @return 分區(qū)號
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // 獲取主題的所有分區(qū)
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        
        // 如果消息值不是MessageEntity類型，拋出異常
        if (!(value instanceof MessageEntity)) {
            throw new InvalidRecordException("消息必須是MessageEntity類型");
        }
        
        MessageEntity message = (MessageEntity) value;
        String businessId = message.getBusinessId();
        
        // 如果業(yè)務(wù)ID為空，使用默認(rèn)分區(qū)策略
        if (StringUtils.isBlank(businessId)) {
            if (keyBytes == null) {
                // 使用隨機(jī)分區(qū)
                return Utils.toPositive(Utils.murmur2(valueBytes)) % numPartitions;
            } else {
                // 使用key計算分區(qū)
                return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
            }
        }
        
        // 根據(jù)業(yè)務(wù)ID計算分區(qū)，確保相同業(yè)務(wù)ID的消息在同一分區(qū)
        return Math.abs(businessId.hashCode()) % numPartitions;
    }
    
    /**
     * 關(guān)閉分區(qū)器
     */
    @Override
    public void close() {
        // 關(guān)閉資源（如果有的話）
    }
    
    /**
     * 配置分區(qū)器
     *
     * @param configs 配置參數(shù)
     */
    @Override
    public void configure(Map<String, ?> configs) {
        // 讀取配置參數(shù)（如果有的話）
    }
}

package com.jam.service;

import com.jam.config.KafkaConstant;
import com.jam.entity.MessageEntity;
import com.jam.entity.MessageTrace;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import java.time.LocalDateTime;
import java.util.Objects;
import java.util.UUID;

/**
 * Kafka消息生產(chǎn)者服務(wù)
 * 負(fù)責(zé)向Kafka發(fā)送各種類型的消息
 *
 * @author 果醬
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class KafkaProducerService {
    /**
     * Kafka模板類，提供發(fā)送消息的各種方法
     */
    private final KafkaTemplate<String, MessageEntity> kafkaTemplate;
    
    /**
     * 事務(wù)Kafka模板類，用于發(fā)送事務(wù)消息
     */
    private final KafkaTemplate<String, MessageEntity> transactionalKafkaTemplate;
    
    /**
     * 消息軌跡服務(wù)
     */
    private final MessageTraceService messageTraceService;
    
    /**
     * 發(fā)送普通消息
     *
     * @param topic 主題名稱
     * @param message 消息實(shí)體
     */
    public void sendMessage(String topic, MessageEntity message) {
        // 參數(shù)校驗
        StringUtils.hasText(topic, "主題名稱不能為空");
        Objects.requireNonNull(message, "消息實(shí)體不能為空");
        
        // 確保消息ID和創(chuàng)建時間不為空
        if (StringUtils.isBlank(message.getMessageId())) {
            message.setMessageId(UUID.randomUUID().toString());
        }
        if (message.getCreateTime() == null) {
            message.setCreateTime(LocalDateTime.now());
        }
        
        // 記錄消息發(fā)送前的軌跡
        messageTraceService.recordBeforeSend(message, topic);
        
        log.info("發(fā)送Kafka消息，主題：{}，消息ID：{}，業(yè)務(wù)類型：{}",
                topic, message.getMessageId(), message.getBusinessType());
        
        // 發(fā)送消息
        ListenableFuture<SendResult<String, MessageEntity>> future = 
            kafkaTemplate.send(topic, message.getMessageId(), message);
        
        // 處理發(fā)送結(jié)果
        future.addCallback(new ListenableFutureCallback<>() {
            @Override
            public void onSuccess(SendResult<String, MessageEntity> result) {
                log.info("Kafka消息發(fā)送成功，主題：{}，消息ID：{}，分區(qū)：{}，偏移量：{}",
                        topic, message.getMessageId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
                
                // 記錄消息發(fā)送成功的軌跡
                messageTraceService.recordSendSuccess(message.getMessageId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
            }
            
            @Override
            public void onFailure(Throwable ex) {
                log.error("Kafka消息發(fā)送失敗，主題：{}，消息ID：{}",
                        topic, message.getMessageId(), ex);
                
                // 記錄消息發(fā)送失敗的軌跡
                messageTraceService.recordSendFailure(message.getMessageId(), ex.getMessage());
            }
        });
    }
    
    /**
     * 發(fā)送分區(qū)消息
     *
     * @param message 消息實(shí)體
     */
    public void sendPartitionMessage(MessageEntity message) {
        // 參數(shù)校驗
        Objects.requireNonNull(message, "消息實(shí)體不能為空");
        StringUtils.hasText(message.getBusinessId(), "業(yè)務(wù)ID不能為空");
        
        // 確保消息ID和創(chuàng)建時間不為空
        if (StringUtils.isBlank(message.getMessageId())) {
            message.setMessageId(UUID.randomUUID().toString());
        }
        if (message.getCreateTime() == null) {
            message.setCreateTime(LocalDateTime.now());
        }
        
        String topic = KafkaConstant.PARTITION_TOPIC;
        
        // 記錄消息發(fā)送前的軌跡
        messageTraceService.recordBeforeSend(message, topic);
        
        log.info("發(fā)送Kafka分區(qū)消息，主題：{}，消息ID：{}，業(yè)務(wù)ID：{}，業(yè)務(wù)類型：{}",
                topic, message.getMessageId(), message.getBusinessId(), message.getBusinessType());
        
        // 發(fā)送消息，使用業(yè)務(wù)ID作為key，配合自定義分區(qū)策略
        ListenableFuture<SendResult<String, MessageEntity>> future = 
            kafkaTemplate.send(topic, message.getBusinessId(), message);
        
        // 處理發(fā)送結(jié)果
        future.addCallback(new ListenableFutureCallback<>() {
            @Override
            public void onSuccess(SendResult<String, MessageEntity> result) {
                log.info("Kafka分區(qū)消息發(fā)送成功，主題：{}，消息ID：{}，業(yè)務(wù)ID：{}，分區(qū)：{}，偏移量：{}",
                        topic, message.getMessageId(), message.getBusinessId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
                
                // 記錄消息發(fā)送成功的軌跡
                messageTraceService.recordSendSuccess(message.getMessageId(),
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
            }
            
            @Override
            public void onFailure(Throwable ex) {
                log.error("Kafka分區(qū)消息發(fā)送失敗，主題：{}，消息ID：{}，業(yè)務(wù)ID：{}",
                        topic, message.getMessageId(), message.getBusinessId(), ex);
                
                // 記錄消息發(fā)送失敗的軌跡
                messageTraceService.recordSendFailure(message.getMessageId(), ex.getMessage());
            }
        });
    }
    
    /**
     * 發(fā)送事務(wù)消息
     *
     * @param message 消息實(shí)體
     */
    @Transactional(rollbackFor = Exception.class)
    public void sendTransactionalMessage(MessageEntity message) {
        // 參數(shù)校驗
        Objects.requireNonNull(message, "消息實(shí)體不能為空");
        
        // 確保消息ID和創(chuàng)建時間不為空
        if (StringUtils.isBlank(message.getMessageId())) {
            message.setMessageId(UUID.randomUUID().toString());
        }
        if (message.getCreateTime() == null) {
            message.setCreateTime(LocalDateTime.now());
        }
        
        String topic = KafkaConstant.TRANSACTIONAL_TOPIC;
        
        // 記錄消息發(fā)送前的軌跡
        messageTraceService.recordBeforeSend(message, topic);
        
        log.info("發(fā)送Kafka事務(wù)消息，主題：{}，消息ID：{}，業(yè)務(wù)類型：{}",
                topic, message.getMessageId(), message.getBusinessType());
        
        // 開始事務(wù)
        transactionalKafkaTemplate.executeInTransaction(kafkaOperations -> {
            // 發(fā)送消息
            SendResult<String, MessageEntity> result = kafkaOperations.send(topic, message.getMessageId(), message).get();
            
            log.info("Kafka事務(wù)消息發(fā)送成功，主題：{}，消息ID：{}，分區(qū)：{}，偏移量：{}",
                    topic, message.getMessageId(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());
            
            // 記錄消息發(fā)送成功的軌跡
            messageTraceService.recordSendSuccess(message.getMessageId(),
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());
            
            // 這里可以添加數(shù)據(jù)庫操作等其他事務(wù)操作
            
            return result;
        });
    }
    
    /**
     * 創(chuàng)建消息實(shí)體
     *
     * @param content 消息內(nèi)容
     * @param businessType 業(yè)務(wù)類型
     * @param businessId 業(yè)務(wù)ID
     * @param extra 額外信息
     * @return 消息實(shí)體
     */
    public MessageEntity createMessageEntity(String content, String businessType, String businessId, String extra) {
        MessageEntity message = new MessageEntity();
        message.setMessageId(UUID.randomUUID().toString());
        message.setContent(content);
        message.setBusinessType(businessType);
        message.setBusinessId(businessId);
        message.setCreateTime(LocalDateTime.now());
        message.setExtra(extra);
        return message;
    }
}

package com.jam.service;

import com.jam.config.KafkaConstant;
import com.jam.entity.MessageEntity;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.Objects;

/**
 * Kafka消息消費(fèi)者服務(wù)
 * 負(fù)責(zé)從Kafka接收并處理消息
 *
 * @author 果醬
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class KafkaConsumerService {
    /**
     * 消息軌跡服務(wù)
     */
    private final MessageTraceService messageTraceService;
    
    /**
     * 消費(fèi)普通消息
     *
     * @param record 消息記錄
     * @param acknowledgment 確認(rèn)對象
     * @param topic 主題名稱
     * @param partition 分區(qū)號
     * @param offset 偏移量
     */
    @KafkaListener(topics = KafkaConstant.NORMAL_TOPIC, groupId = KafkaConstant.NORMAL_CONSUMER_GROUP)
    public void consumeNormalMessage(ConsumerRecord<String, MessageEntity> record,
                                    Acknowledgment acknowledgment,
                                    @Header("kafka_receivedTopic") String topic,
                                    @Header("kafka_receivedPartitionId") int partition,
                                    @Header("kafka_offset") long offset) {
        MessageEntity message = record.value();
        Objects.requireNonNull(message, "消息內(nèi)容不能為空");
        
        log.info("接收到普通消息，主題：{}，分區(qū)：{}，偏移量：{}，消息ID：{}，業(yè)務(wù)類型：{}",
                topic, partition, offset, message.getMessageId(), message.getBusinessType());
        
        try {
            // 處理消息的業(yè)務(wù)邏輯
            processMessage(message);
            
            // 記錄消費(fèi)成功軌跡
            messageTraceService.recordConsumeSuccess(message.getMessageId(), partition, offset);
            
            // 手動確認(rèn)消息
            acknowledgment.acknowledge();
            log.info("普通消息處理成功并確認(rèn)，主題：{}，消息ID：{}", topic, message.getMessageId());
        } catch (Exception e) {
            // 記錄消費(fèi)失敗軌跡
            messageTraceService.recordConsumeFailure(message.getMessageId(), partition, offset, e.getMessage());
            
            log.error("普通消息處理失敗，主題：{}，消息ID：{}", topic, message.getMessageId(), e);
            
            // 手動確認(rèn)消息（將失敗消息標(biāo)記為已消費(fèi)，避免無限重試）
            // 如果需要將消息發(fā)送到死信隊列，可以不確認(rèn)并配置死信轉(zhuǎn)發(fā)
            acknowledgment.acknowledge();
        }
    }
    
    /**
     * 消費(fèi)分區(qū)消息
     *
     * @param record 消息記錄
     * @param acknowledgment 確認(rèn)對象
     * @param topic 主題名稱
     * @param partition 分區(qū)號
     * @param offset 偏移量
     */
    @KafkaListener(topics = KafkaConstant.PARTITION_TOPIC, groupId = KafkaConstant.PARTITION_CONSUMER_GROUP)
    public void consumePartitionMessage(ConsumerRecord<String, MessageEntity> record,
                                       Acknowledgment acknowledgment,
                                       @Header("kafka_receivedTopic") String topic,
                                       @Header("kafka_receivedPartitionId") int partition,
                                       @Header("kafka_offset") long offset) {
        MessageEntity message = record.value();
        Objects.requireNonNull(message, "消息內(nèi)容不能為空");
        
        log.info("接收到分區(qū)消息，主題：{}，分區(qū)：{}，偏移量：{}，消息ID：{}，業(yè)務(wù)ID：{}，業(yè)務(wù)類型：{}",
                topic, partition, offset, message.getMessageId(), message.getBusinessId(), message.getBusinessType());
        
        try {
            // 處理消息的業(yè)務(wù)邏輯
            processMessage(message);
            
            // 記錄消費(fèi)成功軌跡
            messageTraceService.recordConsumeSuccess(message.getMessageId(), partition, offset);
            
            // 手動確認(rèn)消息
            acknowledgment.acknowledge();
            log.info("分區(qū)消息處理成功并確認(rèn)，主題：{}，消息ID：{}", topic, message.getMessageId());
        } catch (Exception e) {
            // 記錄消費(fèi)失敗軌跡
            messageTraceService.recordConsumeFailure(message.getMessageId(), partition, offset, e.getMessage());
            
            log.error("分區(qū)消息處理失敗，主題：{}，消息ID：{}", topic, message.getMessageId(), e);
            acknowledgment.acknowledge();
        }
    }
    
    /**
     * 消費(fèi)事務(wù)消息
     *
     * @param record 消息記錄
     * @param acknowledgment 確認(rèn)對象
     * @param topic 主題名稱
     * @param partition 分區(qū)號
     * @param offset 偏移量
     */
    @Transactional(rollbackFor = Exception.class)
    @KafkaListener(topics = KafkaConstant.TRANSACTIONAL_TOPIC, groupId = KafkaConstant.TRANSACTIONAL_CONSUMER_GROUP)
    public void consumeTransactionalMessage(ConsumerRecord<String, MessageEntity> record,
                                          Acknowledgment acknowledgment,
                                          @Header("kafka_receivedTopic") String topic,
                                          @Header("kafka_receivedPartitionId") int partition,
                                          @Header("kafka_offset") long offset) {
        MessageEntity message = record.value();
        Objects.requireNonNull(message, "消息內(nèi)容不能為空");
        
        log.info("接收到事務(wù)消息，主題：{}，分區(qū)：{}，偏移量：{}，消息ID：{}，業(yè)務(wù)類型：{}",
                topic, partition, offset, message.getMessageId(), message.getBusinessType());
        
        try {
            // 處理消息的業(yè)務(wù)邏輯
            processMessage(message);
            
            // 這里可以添加數(shù)據(jù)庫操作等其他事務(wù)操作
            
            // 記錄消費(fèi)成功軌跡
            messageTraceService.recordConsumeSuccess(message.getMessageId(), partition, offset);
            
            // 手動確認(rèn)消息
            acknowledgment.acknowledge();
            log.info("事務(wù)消息處理成功并確認(rèn)，主題：{}，消息ID：{}", topic, message.getMessageId());
        } catch (Exception e) {
            // 記錄消費(fèi)失敗軌跡
            messageTraceService.recordConsumeFailure(message.getMessageId(), partition, offset, e.getMessage());
            
            log.error("事務(wù)消息處理失敗，主題：{}，消息ID：{}", topic, message.getMessageId(), e);
            // 事務(wù)會回滾，消息不會被確認(rèn)，將被重新消費(fèi)
        }
    }
    
    /**
     * 消費(fèi)死信消息
     *
     * @param record 消息記錄
     * @param acknowledgment 確認(rèn)對象
     * @param topic 主題名稱
     * @param partition 分區(qū)號
     * @param offset 偏移量
     */
    @KafkaListener(topics = KafkaConstant.DEAD_LETTER_TOPIC, groupId = KafkaConstant.DEAD_LETTER_CONSUMER_GROUP)
    public void consumeDeadLetterMessage(ConsumerRecord<String, MessageEntity> record,
                                        Acknowledgment acknowledgment,
                                        @Header("kafka_receivedTopic") String topic,
                                        @Header("kafka_receivedPartitionId") int partition,
                                        @Header("kafka_offset") long offset) {
        MessageEntity message = record.value();
        Objects.requireNonNull(message, "消息內(nèi)容不能為空");
        
        log.error("接收到死信消息，主題：{}，分區(qū)：{}，偏移量：{}，消息ID：{}，業(yè)務(wù)類型：{}",
                topic, partition, offset, message.getMessageId(), message.getBusinessType());
        
        try {
            // 處理死信消息的業(yè)務(wù)邏輯，通常需要人工干預(yù)
            processDeadLetterMessage(message);
            
            // 記錄消費(fèi)成功軌跡
            messageTraceService.recordConsumeSuccess(message.getMessageId(), partition, offset);
            
            // 手動確認(rèn)消息
            acknowledgment.acknowledge();
            log.info("死信消息處理成功并確認(rèn)，主題：{}，消息ID：{}", topic, message.getMessageId());
        } catch (Exception e) {
            // 記錄消費(fèi)失敗軌跡
            messageTraceService.recordConsumeFailure(message.getMessageId(), partition, offset, e.getMessage());
            
            log.error("死信消息處理失敗，主題：{}，消息ID：{}", topic, message.getMessageId(), e);
            acknowledgment.acknowledge();
        }
    }
    
    /**
     * 處理消息的業(yè)務(wù)邏輯
     *
     * @param message 要處理的消息
     */
    private void processMessage(MessageEntity message) {
        // 根據(jù)業(yè)務(wù)類型處理不同的消息
        String businessType = message.getBusinessType();
        if ("ORDER_CREATE".equals(businessType)) {
            // 處理訂單創(chuàng)建消息
            processOrderCreateMessage(message);
        } else if ("USER_REGISTER".equals(businessType)) {
            // 處理用戶注冊消息
            processUserRegisterMessage(message);
        } else {
            // 處理未知類型消息
            log.warn("收到未知類型的消息，消息ID：{}，業(yè)務(wù)類型：{}",
                    message.getMessageId(), businessType);
        }
    }
    
    /**
     * 處理死信消息
     *
     * @param message 死信消息
     */
    private void processDeadLetterMessage(MessageEntity message) {
        log.info("處理死信消息，消息ID：{}，內(nèi)容：{}",
                message.getMessageId(), message.getContent());
        // 實(shí)際業(yè)務(wù)處理邏輯，如記錄到數(shù)據(jù)庫等待人工處理
    }
    
    /**
     * 處理訂單創(chuàng)建消息
     *
     * @param message 訂單創(chuàng)建消息
     */
    private void processOrderCreateMessage(MessageEntity message) {
        log.info("處理訂單創(chuàng)建消息，消息ID：{}，訂單信息：{}",
                message.getMessageId(), message.getContent());
        // 實(shí)際業(yè)務(wù)處理邏輯...
    }
    
    /**
     * 處理用戶注冊消息
     *
     * @param message 用戶注冊消息
     */
    private void processUserRegisterMessage(MessageEntity message) {
        log.info("處理用戶注冊消息，消息ID：{}，用戶信息：{}",
                message.getMessageId(), message.getContent());
        // 實(shí)際業(yè)務(wù)處理邏輯...
    }
}

package com.jam.service;

import com.jam.entity.MessageEntity;
import com.jam.entity.MessageTrace;
import com.jam.mapper.MessageTraceMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.util.Objects;

/**
 * 消息軌跡服務(wù)
 * 記錄消息的發(fā)送和消費(fèi)軌跡
 *
 * @author 果醬
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class MessageTraceService {
    private final MessageTraceMapper messageTraceMapper;
    
    /**
     * 記錄消息發(fā)送前的軌跡
     *
     * @param message 消息實(shí)體
     * @param topic 主題
     * @return 消息軌跡ID
     */
    @Transactional(rollbackFor = Exception.class)
    public Long recordBeforeSend(MessageEntity message, String topic) {
        Objects.requireNonNull(message, "消息實(shí)體不能為空");
        StringUtils.hasText(message.getMessageId(), "消息ID不能為空");
        StringUtils.hasText(topic, "主題不能為空");
        
        MessageTrace trace = new MessageTrace();
        trace.setMessageId(message.getMessageId());
        trace.setTopic(topic);
        trace.setBusinessType(message.getBusinessType());
        trace.setBusinessId(message.getBusinessId());
        trace.setContent(message.getContent());
        trace.setSendStatus(0); // 待發(fā)送
        trace.setCreateTime(LocalDateTime.now());
        trace.setUpdateTime(LocalDateTime.now());
        
        messageTraceMapper.insert(trace);
        log.info("記錄消息發(fā)送前軌跡，消息ID：{}，軌跡ID：{}", message.getMessageId(), trace.getId());
        return trace.getId();
    }
    
    /**
     * 記錄消息發(fā)送成功的軌跡
     *
     * @param messageId 消息ID
     * @param partition 分區(qū)
     * @param offset 偏移量
     */
    @Transactional(rollbackFor = Exception.class)
    public void recordSendSuccess(String messageId, int partition, long offset) {
        StringUtils.hasText(messageId, "消息ID不能為空");
        
        MessageTrace trace = messageTraceMapper.selectByMessageId(messageId);
        if (trace == null) {
            log.warn("未找到消息軌跡，消息ID：{}", messageId);
            return;
        }
        
        trace.setSendTime(LocalDateTime.now());
        trace.setSendStatus(1); // 發(fā)送成功
        trace.setPartition(partition);
        trace.setOffset(offset);
        trace.setUpdateTime(LocalDateTime.now());
        
        messageTraceMapper.updateById(trace);
        log.info("記錄消息發(fā)送成功軌跡，消息ID：{}", messageId);
    }
    
    /**
     * 記錄消息發(fā)送失敗的軌跡
     *
     * @param messageId 消息ID
     * @param errorMsg 錯誤信息
     */
    @Transactional(rollbackFor = Exception.class)
    public void recordSendFailure(String messageId, String errorMsg) {
        StringUtils.hasText(messageId, "消息ID不能為空");
        StringUtils.hasText(errorMsg, "錯誤信息不能為空");
        
        MessageTrace trace = messageTraceMapper.selectByMessageId(messageId);
        if (trace == null) {
            log.warn("未找到消息軌跡，消息ID：{}", messageId);
            return;
        }
        
        trace.setSendTime(LocalDateTime.now());
        trace.setSendStatus(2); // 發(fā)送失敗
        trace.setSendErrorMsg(errorMsg);
        trace.setUpdateTime(LocalDateTime.now());
        
        messageTraceMapper.updateById(trace);
        log.info("記錄消息發(fā)送失敗軌跡，消息ID：{}", messageId);
    }
    
    /**
     * 記錄消息消費(fèi)成功的軌跡
     *
     * @param messageId 消息ID
     * @param partition 分區(qū)
     * @param offset 偏移量
     */
    @Transactional(rollbackFor = Exception.class)
    public void recordConsumeSuccess(String messageId, int partition, long offset) {
        StringUtils.hasText(messageId, "消息ID不能為空");
        
        MessageTrace trace = messageTraceMapper.selectByMessageId(messageId);
        if (trace == null) {
            log.warn("未找到消息軌跡，消息ID：{}", messageId);
            return;
        }
        
        trace.setConsumeTime(LocalDateTime.now());
        trace.setConsumeStatus(1); // 消費(fèi)成功
        trace.setConsumePartition(partition);
        trace.setConsumeOffset(offset);
        trace.setUpdateTime(LocalDateTime.now());
        
        messageTraceMapper.updateById(trace);
        log.info("記錄消息消費(fèi)成功軌跡，消息ID：{}", messageId);
    }
    
    /**
     * 記錄消息消費(fèi)失敗的軌跡
     *
     * @param messageId 消息ID
     * @param partition 分區(qū)
     * @param offset 偏移量
     * @param errorMsg 錯誤信息
     */
    @Transactional(rollbackFor = Exception.class)
    public void recordConsumeFailure(String messageId, int partition, long offset, String errorMsg) {
        StringUtils.hasText(messageId, "消息ID不能為空");
        StringUtils.hasText(errorMsg, "錯誤信息不能為空");
        
        MessageTrace trace = messageTraceMapper.selectByMessageId(messageId);
        if (trace == null) {
            log.warn("未找到消息軌跡，消息ID：{}", messageId);
            return;
        }
        
        trace.setConsumeTime(LocalDateTime.now());
        trace.setConsumeStatus(2); // 消費(fèi)失敗
        trace.setConsumePartition(partition);
        trace.setConsumeOffset(offset);
        trace.setConsumeErrorMsg(errorMsg);
        trace.setUpdateTime(LocalDateTime.now());
        
        messageTraceMapper.updateById(trace);
        log.info("記錄消息消費(fèi)失敗軌跡，消息ID：{}", messageId);
    }
}

package com.jam.controller;

import com.jam.config.KafkaConstant;
import com.jam.entity.MessageEntity;
import com.jam.service.KafkaProducerService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * Kafka消息測試控制器
 * 提供API接口用于測試Kafka消息發(fā)送功能
 *
 * @author 果醬
 */
@Slf4j
@RestController
@RequestMapping("/api/kafka")
@RequiredArgsConstructor
@Tag(name = "Kafka消息測試接口", description = "用于測試Kafka消息發(fā)送的API接口")
public class KafkaMessageController {
    /**
     * Kafka消息生產(chǎn)者服務(wù)
     */
    private final KafkaProducerService kafkaProducerService;
    
    /**
     * 發(fā)送普通消息
     *
     * @param content 消息內(nèi)容
     * @param businessType 業(yè)務(wù)類型
     * @param businessId 業(yè)務(wù)ID
     * @param extra 額外信息
     * @return 響應(yīng)信息
     */
    @PostMapping("/normal")
    @Operation(summary = "發(fā)送普通消息", description = "發(fā)送到普通主題的消息")
    public ResponseEntity<String> sendNormalMessage(
            @Parameter(description = "消息內(nèi)容", required = true)
            @RequestParam String content,
            
            @Parameter(description = "業(yè)務(wù)類型")
            @RequestParam(required = false) String businessType,
            
            @Parameter(description = "業(yè)務(wù)ID")
            @RequestParam(required = false) String businessId,
            
            @Parameter(description = "額外信息")
            @RequestParam(required = false) String extra) {
        
        log.info("接收到發(fā)送普通消息請求");
        MessageEntity message = kafkaProducerService.createMessageEntity(content, businessType, businessId, extra);
        kafkaProducerService.sendMessage(KafkaConstant.NORMAL_TOPIC, message);
        return ResponseEntity.ok("普通消息發(fā)送成功，消息ID：" + message.getMessageId());
    }
    
    /**
     * 發(fā)送分區(qū)消息
     *
     * @param content 消息內(nèi)容
     * @param businessType 業(yè)務(wù)類型
     * @param businessId 業(yè)務(wù)ID（用于分區(qū)）
     * @param extra 額外信息
     * @return 響應(yīng)信息
     */
    @PostMapping("/partition")
    @Operation(summary = "發(fā)送分區(qū)消息", description = "發(fā)送到分區(qū)主題的消息，相同業(yè)務(wù)ID的消息會被發(fā)送到同一分區(qū)")
    public ResponseEntity<String> sendPartitionMessage(
            @Parameter(description = "消息內(nèi)容", required = true)
            @RequestParam String content,
            
            @Parameter(description = "業(yè)務(wù)類型")
            @RequestParam(required = false) String businessType,
            
            @Parameter(description = "業(yè)務(wù)ID（用于分區(qū)）", required = true)
            @RequestParam String businessId,
            
            @Parameter(description = "額外信息")
            @RequestParam(required = false) String extra) {
        
        log.info("接收到發(fā)送分區(qū)消息請求，業(yè)務(wù)ID：{}", businessId);
        MessageEntity message = kafkaProducerService.createMessageEntity(content, businessType, businessId, extra);
        kafkaProducerService.sendPartitionMessage(message);
        return ResponseEntity.ok("分區(qū)消息發(fā)送成功，消息ID：" + message.getMessageId());
    }
    
    /**
     * 發(fā)送事務(wù)消息
     *
     * @param content 消息內(nèi)容
     * @param businessType 業(yè)務(wù)類型
     * @param businessId 業(yè)務(wù)ID
     * @param extra 額外信息
     * @return 響應(yīng)信息
     */
    @PostMapping("/transactional")
    @Operation(summary = "發(fā)送事務(wù)消息", description = "發(fā)送到事務(wù)主題的消息，支持事務(wù)特性")
    public ResponseEntity<String> sendTransactionalMessage(
            @Parameter(description = "消息內(nèi)容", required = true)
            @RequestParam String content,
            
            @Parameter(description = "業(yè)務(wù)類型")
            @RequestParam(required = false) String businessType,
            
            @Parameter(description = "業(yè)務(wù)ID")
            @RequestParam(required = false) String businessId,
            
            @Parameter(description = "額外信息")
            @RequestParam(required = false) String extra) {
        
        log.info("接收到發(fā)送事務(wù)消息請求");
        MessageEntity message = kafkaProducerService.createMessageEntity(content, businessType, businessId, extra);
        kafkaProducerService.sendTransactionalMessage(message);
        return ResponseEntity.ok("事務(wù)消息發(fā)送成功，消息ID：" + message.getMessageId());
    }
}

package com.jam;

import io.swagger.v3.oas.annotations.OpenAPIDefinition;
import io.swagger.v3.oas.annotations.info.Info;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * SpringBoot應(yīng)用啟動類
 *
 * @author 果醬
 */
@SpringBootApplication
@MapperScan("com.jam.mapper")
@OpenAPIDefinition(
        info = @Info(
                title = "SpringBoot集成Kafka示例項目",
                version = "1.0",
                description = "SpringBoot集成Kafka的示例項目，包含各種消息發(fā)送和消費(fèi)的示例"
        )
)
public class SpringbootKafkaDemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(SpringbootKafkaDemoApplication.class, args);
    }
}

package com.jam.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import java.time.LocalDateTime;

/**
 * 消息軌跡實(shí)體類
 * 記錄Kafka消息的發(fā)送和消費(fèi)情況
 *
 * @author 果醬
 */
@Data
@TableName("t_message_trace")
public class MessageTrace {
    /**
     * 主鍵ID
     */
    @TableId(type = IdType.AUTO)
    private Long id;
    
    /**
     * 消息ID
     */
    private String messageId;
    
    /**
     * 主題
     */
    private String topic;
    
    /**
     * 分區(qū)
     */
    private Integer partition;
    
    /**
     * 偏移量
     */
    private Long offset;
    
    /**
     * 業(yè)務(wù)類型
     */
    private String businessType;
    
    /**
     * 業(yè)務(wù)ID
     */
    private String businessId;
    
    /**
     * 消息內(nèi)容
     */
    private String content;
    
    /**
     * 發(fā)送時間
     */
    private LocalDateTime sendTime;
    
    /**
     * 發(fā)送狀態(tài)：0-待發(fā)送，1-發(fā)送成功，2-發(fā)送失敗
     */
    private Integer sendStatus;
    
    /**
     * 發(fā)送錯誤信息
     */
    private String sendErrorMsg;
    
    /**
     * 消費(fèi)時間
     */
    private LocalDateTime consumeTime;
    
    /**
     * 消費(fèi)分區(qū)
     */
    private Integer consumePartition;
    
    /**
     * 消費(fèi)偏移量
     */
    private Long consumeOffset;
    
    /**
     * 消費(fèi)狀態(tài)：0-待消費(fèi)，1-消費(fèi)成功，2-消費(fèi)失敗
     */
    private Integer consumeStatus;
    
    /**
     * 消費(fèi)錯誤信息
     */
    private String consumeErrorMsg;
    
    /**
     * 創(chuàng)建時間
     */
    private LocalDateTime createTime;
    
    /**
     * 更新時間
     */
    private LocalDateTime updateTime;
}

package com.jam.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.jam.entity.MessageTrace;
import org.apache.ibatis.annotations.Param;

/**
 * 消息軌跡Mapper
 *
 * @author 果醬
 */
public interface MessageTraceMapper extends BaseMapper<MessageTrace> {
    /**
     * 根據(jù)消息ID查詢消息軌跡
     *
     * @param messageId 消息ID
     * @return 消息軌跡信息
     */
    MessageTrace selectByMessageId(@Param("messageId") String messageId);
}

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.jam.mapper.MessageTraceMapper">
    <select id="selectByMessageId" parameterType="java.lang.String" resultType="com.jam.entity.MessageTrace">
        SELECT * FROM t_message_trace WHERE message_id = #{messageId}
    </select>
</mapper>

CREATE TABLE `t_message_trace` (
  `id` bigint NOT NULL AUTO_INCREMENT COMMENT '主鍵ID',
  `message_id` varchar(64) NOT NULL COMMENT '消息ID',
  `topic` varchar(128) NOT NULL COMMENT '主題',
  `partition` int DEFAULT NULL COMMENT '分區(qū)',
  `offset` bigint DEFAULT NULL COMMENT '偏移量',
  `business_type` varchar(64) DEFAULT NULL COMMENT '業(yè)務(wù)類型',
  `business_id` varchar(64) DEFAULT NULL COMMENT '業(yè)務(wù)ID',
  `content` text COMMENT '消息內(nèi)容',
  `send_time` datetime DEFAULT NULL COMMENT '發(fā)送時間',
  `send_status` tinyint DEFAULT NULL COMMENT '發(fā)送狀態(tài)：0-待發(fā)送，1-發(fā)送成功，2-發(fā)送失敗',
  `send_error_msg` text COMMENT '發(fā)送錯誤信息',
  `consume_time` datetime DEFAULT NULL COMMENT '消費(fèi)時間',
  `consume_partition` int DEFAULT NULL COMMENT '消費(fèi)分區(qū)',
  `consume_offset` bigint DEFAULT NULL COMMENT '消費(fèi)偏移量',
  `consume_status` tinyint DEFAULT NULL COMMENT '消費(fèi)狀態(tài)：0-待消費(fèi)，1-消費(fèi)成功，2-消費(fèi)失敗',
  `consume_error_msg` text COMMENT '消費(fèi)錯誤信息',
  `create_time` datetime NOT NULL COMMENT '創(chuàng)建時間',
  `update_time` datetime NOT NULL COMMENT '更新時間',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_message_id` (`message_id`),
  KEY `idx_topic` (`topic`),
  KEY `idx_business_type` (`business_type`),
  KEY `idx_business_id` (`business_id`),
  KEY `idx_send_status` (`send_status`),
  KEY `idx_consume_status` (`consume_status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Kafka消息軌跡表';

# 發(fā)送普通消息
curl -X POST "http://localhost:8081/api/kafka/normal?content=Hello Kafka&businessType=TEST"

# 發(fā)送分區(qū)消息
curl -X POST "http://localhost:8081/api/kafka/partition?content=Hello Partition&businessType=TEST&businessId=BUS123456"

# 發(fā)送事務(wù)消息
curl -X POST "http://localhost:8081/api/kafka/transactional?content=Hello Transaction&businessType=TEST"

/**
 * 轉(zhuǎn)發(fā)消息到死信隊列
 *
 * @param message 消息實(shí)體
 * @param topic 原主題
 * @param partition 原分區(qū)
 * @param offset 原偏移量
 * @param errorMsg 錯誤信息
 */
public void forwardToDeadLetterQueue(MessageEntity message, String topic, int partition, long offset, String errorMsg) {
    Objects.requireNonNull(message, "消息實(shí)體不能為空");
    StringUtils.hasText(topic, "主題不能為空");
    StringUtils.hasText(errorMsg, "錯誤信息不能為空");
    
    log.warn("將消息轉(zhuǎn)發(fā)到死信隊列，原主題：{}，消息ID：{}，錯誤信息：{}",
            topic, message.getMessageId(), errorMsg);
    
    // 創(chuàng)建死信消息，添加原消息的元數(shù)據(jù)
    MessageEntity deadLetterMessage = new MessageEntity();
    deadLetterMessage.setMessageId(UUID.randomUUID().toString());
    deadLetterMessage.setContent(JSON.toJSONString(message));
    deadLetterMessage.setBusinessType("DEAD_LETTER");
    deadLetterMessage.setBusinessId(message.getMessageId());
    deadLetterMessage.setCreateTime(LocalDateTime.now());
    deadLetterMessage.setExtra(String.format("原主題：%s，原分區(qū)：%d，原偏移量：%d，錯誤信息：%s",
            topic, partition, offset, errorMsg));
    
    // 發(fā)送到死信主題
    kafkaTemplate.send(KafkaConstant.DEAD_LETTER_TOPIC, deadLetterMessage.getMessageId(), deadLetterMessage);
}

/**
 * 處理消息（冪等性保證）
 *
 * @param message 消息實(shí)體
 */
@Transactional(rollbackFor = Exception.class)
public void processMessageWithIdempotency(MessageEntity message) {
    String messageId = message.getMessageId();
    String businessType = message.getBusinessType();
    
    // 檢查消息是否已經(jīng)處理過
    MessageTrace trace = messageTraceMapper.selectByMessageId(messageId);
    if (trace != null && trace.getConsumeStatus() == 1) {
        log.info("消息已經(jīng)處理過，消息ID：{}", messageId);
        return;
    }
    
    // 根據(jù)業(yè)務(wù)類型處理不同的消息
    if ("ORDER_CREATE".equals(businessType)) {
        // 處理訂單創(chuàng)建消息，使用訂單號作為唯一鍵
        String orderNo = message.getExtra();
        // 檢查訂單是否已經(jīng)處理
        Order order = orderMapper.selectByOrderNo(orderNo);
        if (order != null) {
            log.info("訂單已經(jīng)處理過，訂單號：{}", orderNo);
            return;
        }
        // 處理訂單業(yè)務(wù)邏輯
        // ...
    } else if ("USER_REGISTER".equals(businessType)) {
        // 處理用戶注冊消息，使用用戶ID作為唯一鍵
        // ...
    }
}

/**
 * 使用Redis分布式鎖保證冪等性
 *
 * @param message 消息實(shí)體
 */
public void processMessageWithRedisLock(MessageEntity message) {
    String messageId = message.getMessageId();
    String lockKey = "kafka:message:process:" + messageId;
    
    // 獲取分布式鎖，設(shè)置5分鐘過期時間
    Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, "1", 5, TimeUnit.MINUTES);
    
    if (Boolean.TRUE.equals(locked)) {
        try {
            // 檢查消息是否已經(jīng)處理過
            MessageTrace trace = messageTraceMapper.selectByMessageId(messageId);
            if (trace != null && trace.getConsumeStatus() == 1) {
                log.info("消息已經(jīng)處理過，消息ID：{}", messageId);
                return;
            }
            
            // 處理消息業(yè)務(wù)邏輯
            processMessage(message);
            
        } finally {
            // 釋放鎖
            redisTemplate.delete(lockKey);
        }
    } else {
        log.info("消息正在處理中，消息ID：{}", messageId);
    }
}