graph TB
    A[啟動(dòng)類] --> B[@SpringBootApplication]
    B --> C[@EnableAutoConfiguration]
    C --> D[spring.factories]
    D --> E[自動(dòng)配置類]
    E --> F[條件注解檢查]
    F --> G[創(chuàng)建Bean]
    G --> H[完成整合]
    style E fill:#bbdefb,stroke:#333
    style F fill:#c8e6c9,stroke:#333

@Configuration
@ConditionalOnClass(RedisTemplate.class)  // 類路徑存在時(shí)生效
@ConditionalOnProperty(prefix = "spring.redis", name = "enabled", matchIfMissing = true)
@EnableConfigurationProperties(RedisProperties.class)  // 綁定配置
public class RedisAutoConfiguration {
    @Bean
    @ConditionalOnMissingBean  // 容器中不存在時(shí)創(chuàng)建
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);
        return template;
    }
}

<!-- pom.xml -->
<dependencies>
    <!-- Redis Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <!-- 連接池 -->
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-pool2</artifactId>
    </dependency>
    <!-- JSON 序列化 -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
</dependencies>

# application.yml
spring:
  redis:
    # 連接配置
    host: localhost
    port: 6379
    password: 123456
    database: 0
    timeout: 2000ms
    # 連接池配置
    lettuce:
      pool:
        max-active: 20      # 最大連接數(shù)
        max-idle: 10       # 最大空閑連接
        min-idle: 5        # 最小空閑連接
        max-wait: 1000ms   # 獲取連接最大等待時(shí)間
    # 集群配置（可選）
    # cluster:
    #   nodes: 
    #     - 192.168.1.101:6379
    #     - 192.168.1.102:6379

@Configuration
@Slf4j
public class RedisConfig {
    @Autowired
    private RedisConnectionFactory redisConnectionFactory;
    /**
     * 自定義 RedisTemplate，解決序列化問題
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate() {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);
        // 使用 Jackson2JsonRedisSerializer 替換默認(rèn)序列化
        Jackson2JsonRedisSerializer<Object> serializer = 
            new Jackson2JsonRedisSerializer<>(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.activateDefaultTyping(
            LazyIterator.class, ObjectMapper.DefaultTyping.NON_FINAL);
        serializer.setObjectMapper(objectMapper);
        // 設(shè)置 key 和 value 的序列化規(guī)則
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(serializer);
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(serializer);
        template.afterPropertiesSet();
        log.info("RedisTemplate 配置完成");
        return template;
    }
    /**
     * 緩存管理器配置
     */
    @Bean
    public RedisCacheManager cacheManager(RedisConnectionFactory factory) {
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
            .entryTtl(Duration.ofMinutes(30))  // 默認(rèn)緩存30分鐘
            .disableCachingNullValues()         // 不緩存null值
            .serializeKeysWith(RedisSerializationContext.SerializationPair
                .fromSerializer(new StringRedisSerializer()))
            .serializeValuesWith(RedisSerializationContext.SerializationPair
                .fromSerializer(new GenericJackson2JsonRedisSerializer()));
        return RedisCacheManager.builder(factory)
            .cacheDefaults(config)
            .build();
    }
}

@Service
@Slf4j
public class RedisOperationService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 字符串操作
     */
    public void stringOperations() {
        // 1. 設(shè)置值
        redisTemplate.opsForValue().set("user:1001:name", "張三");
        redisTemplate.opsForValue().set("user:1001:age", 25, Duration.ofMinutes(30));
        // 2. 獲取值
        String name = (String) redisTemplate.opsForValue().get("user:1001:name");
        Integer age = (Integer) redisTemplate.opsForValue().get("user:1001:age");
        log.info("用戶信息 - 姓名: {}, 年齡: {}", name, age);
        // 3. 原子操作
        Long increment = redisTemplate.opsForValue().increment("counter", 1);
        log.info("計(jì)數(shù)器值: {}", increment);
    }
    /**
     * Hash 操作
     */
    public void hashOperations() {
        String key = "user:1001:profile";
        // 1. 設(shè)置Hash字段
        Map<String, Object> userMap = new HashMap<>();
        userMap.put("name", "李四");
        userMap.put("age", 30);
        userMap.put("email", "lisi@example.com");
        redisTemplate.opsForHash().putAll(key, userMap);
        // 2. 獲取單個(gè)字段
        String name = (String) redisTemplate.opsForHash().get(key, "name");
        log.info("Hash字段 name: {}", name);
        // 3. 獲取所有字段
        Map<Object, Object> entries = redisTemplate.opsForHash().entries(key);
        log.info("完整用戶信息: {}", entries);
    }
    /**
     * List 操作
     */
    public void listOperations() {
        String key = "recent:users";
        // 1. 從左端插入
        redisTemplate.opsForList().leftPush(key, "user1");
        redisTemplate.opsForList().leftPush(key, "user2");
        redisTemplate.opsForList().leftPush(key, "user3");
        // 2. 獲取范圍
        List<Object> recentUsers = redisTemplate.opsForList().range(key, 0, 2);
        log.info("最近訪問用戶: {}", recentUsers);
    }
    /**
     * Set 操作
     */
    public void setOperations() {
        String key = "user:tags:1001";
        // 1. 添加元素
        redisTemplate.opsForSet().add(key, "vip", "active", "new");
        // 2. 判斷元素是否存在
        Boolean isVip = redisTemplate.opsForSet().isMember(key, "vip");
        log.info("是否是VIP: {}", isVip);
        // 3. 獲取所有元素
        Set<Object> tags = redisTemplate.opsForSet().members(key);
        log.info("用戶標(biāo)簽: {}", tags);
    }
}

@Service
@Slf4j
public class UserService {
    /**
     * 緩存用戶信息
     */
    @Cacheable(value = "users", key = "#userId", unless = "#result == null")
    public User getUserById(Long userId) {
        log.info("查詢數(shù)據(jù)庫獲取用戶: {}", userId);
        // 模擬數(shù)據(jù)庫查詢
        return findUserFromDB(userId);
    }
    /**
     * 更新緩存
     */
    @CachePut(value = "users", key = "#user.id")
    public User updateUser(User user) {
        log.info("更新用戶信息: {}", user.getId());
        // 模擬數(shù)據(jù)庫更新
        return updateUserInDB(user);
    }
    /**
     * 刪除緩存
     */
    @CacheEvict(value = "users", key = "#userId")
    public void deleteUser(Long userId) {
        log.info("刪除用戶: {}", userId);
        // 模擬數(shù)據(jù)庫刪除
        deleteUserFromDB(userId);
    }
    /**
     * 復(fù)雜緩存配置
     */
    @Caching(
        cacheable = {
            @Cacheable(value = "user_detail", key = "#userId")
        },
        put = {
            @CachePut(value = "user_profile", key = "#userId")
        }
    )
    public User getUserDetail(Long userId) {
        log.info("獲取用戶詳情: {}", userId);
        return findUserDetailFromDB(userId);
    }
    // 模擬數(shù)據(jù)庫操作
    private User findUserFromDB(Long userId) {
        // 實(shí)際項(xiàng)目中這里會(huì)查詢數(shù)據(jù)庫
        return new User(userId, "用戶" + userId, "user" + userId + "@example.com");
    }
    private User updateUserInDB(User user) {
        return user;
    }
    private void deleteUserFromDB(Long userId) {
        // 刪除操作
    }
    private User findUserDetailFromDB(Long userId) {
        return new User(userId, "詳情用戶" + userId, "detail" + userId + "@example.com");
    }
    @Data
    @AllArgsConstructor
    public static class User {
        private Long id;
        private String name;
        private String email;
    }
}

@Component
public class RedisHealthIndicator implements HealthIndicator {
    @Autowired
    private RedisConnectionFactory redisConnectionFactory;
    @Override
    public Health health() {
        try {
            // 測(cè)試Redis連接
            RedisConnection connection = redisConnectionFactory.getConnection();
            try {
                String result = connection.ping();
                if ("PONG".equals(result)) {
                    return Health.up()
                        .withDetail("server", "redis")
                        .withDetail("version", getRedisVersion(connection))
                        .withDetail("status", "connected")
                        .build();
                } else {
                    return Health.down()
                        .withDetail("error", "Unexpected ping response: " + result)
                        .build();
                }
            } finally {
                connection.close();
            }
        } catch (Exception e) {
            return Health.down(e)
                .withDetail("error", "Redis connection failed: " + e.getMessage())
                .build();
        }
    }
    private String getRedisVersion(RedisConnection connection) {
        try {
            Properties info = connection.info("server");
            return info.getProperty("redis_version");
        } catch (Exception e) {
            return "unknown";
        }
    }
}

<!-- pom.xml -->
<dependencies>
    <!-- MyBatis Starter -->
    <dependency>
        <groupId>org.mybatis.spring.boot</groupId>
        <artifactId>mybatis-spring-boot-starter</artifactId>
        <version>2.2.2</version>
    </dependency>
    <!-- MySQL 驅(qū)動(dòng) -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <scope>runtime</scope>
    </dependency>
    <!-- 數(shù)據(jù)源 -->
    <dependency>
        <groupId>com.zaxxer</groupId>
        <artifactId>HikariCP</artifactId>
    </dependency>
</dependencies>

# application.yml
spring:
  datasource:
    # 數(shù)據(jù)源配置
    url: jdbc:mysql://localhost:3306/spring_boot_demo?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai
    username: root
    password: 123456
    driver-class-name: com.mysql.cj.jdbc.Driver
    # Hikari 連接池配置
    hikari:
      maximum-pool-size: 20
      minimum-idle: 5
      connection-timeout: 30000
      idle-timeout: 300000
      max-lifetime: 1200000
# MyBatis 配置
mybatis:
  #  mapper 文件位置
  mapper-locations: classpath:mapper/*.xml
  # 別名掃描包
  type-aliases-package: com.example.entity
  # 開啟駝峰命名轉(zhuǎn)換
  configuration:
    map-underscore-to-camel-case: true
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
  # 全局配置
  global-config:
    db-config:
      id-type: auto
      logic-delete-field: deleted  # 邏輯刪除字段
      logic-delete-value: 1        # 邏輯已刪除值
      logic-not-delete-value: 0    # 邏輯未刪除值

@Data
@NoArgsConstructor
@AllArgsConstructor
@TableName("user")  // MyBatis-Plus 注解，標(biāo)準(zhǔn) MyBatis 可省略
public class User {
    @TableId(type = IdType.AUTO)  // 主鍵自增
    private Long id;
    private String name;
    private String email;
    private Integer age;
    @TableField(fill = FieldFill.INSERT)  // 插入時(shí)自動(dòng)填充
    private LocalDateTime createTime;
    @TableField(fill = FieldFill.INSERT_UPDATE)  // 插入和更新時(shí)填充
    private LocalDateTime updateTime;
    @TableLogic  // 邏輯刪除標(biāo)識(shí)
    private Integer deleted;
}

@Mapper
@Repository
public interface UserMapper {
    /**
     * 根據(jù)ID查詢用戶
     */
    @Select("SELECT * FROM user WHERE id = #{id} AND deleted = 0")
    User selectById(Long id);
    /**
     * 查詢所有用戶
     */
    @Select("SELECT * FROM user WHERE deleted = 0 ORDER BY create_time DESC")
    List<User> selectAll();
    /**
     * 插入用戶
     */
    @Insert("INSERT INTO user(name, email, age, create_time, update_time) " +
            "VALUES(#{name}, #{email}, #{age}, NOW(), NOW())")
    @Options(useGeneratedKeys = true, keyProperty = "id")
    int insert(User user);
    /**
     * 更新用戶
     */
    @Update("UPDATE user SET name=#{name}, email=#{email}, age=#{age}, " +
            "update_time=NOW() WHERE id=#{id} AND deleted = 0")
    int update(User user);
    /**
     * 邏輯刪除用戶
     */
    @Update("UPDATE user SET deleted=1, update_time=NOW() WHERE id=#{id}")
    int deleteById(Long id);
    /**
     * 根據(jù)郵箱查詢用戶
     */
    @Select("SELECT * FROM user WHERE email = #{email} AND deleted = 0")
    User selectByEmail(String email);
    /**
     * 分頁查詢用戶
     */
    @Select("SELECT * FROM user WHERE deleted = 0 ORDER BY create_time DESC " +
            "LIMIT #{offset}, #{pageSize}")
    List<User> selectByPage(@Param("offset") int offset, 
                           @Param("pageSize") int pageSize);
}

<?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.example.mapper.UserMapper">
    <!-- 自定義結(jié)果映射 -->
    <resultMap id="UserResultMap" type="User">
        <id column="id" property="id" />
        <result column="name" property="name" />
        <result column="email" property="email" />
        <result column="age" property="age" />
        <result column="create_time" property="createTime" />
        <result column="update_time" property="updateTime" />
        <result column="deleted" property="deleted" />
    </resultMap>
    <!-- 復(fù)雜查詢：根據(jù)條件動(dòng)態(tài)查詢用戶 -->
    <select id="selectByCondition" parameterType="map" resultMap="UserResultMap">
        SELECT * FROM user 
        WHERE deleted = 0
        <if test="name != null and name != ''">
            AND name LIKE CONCAT('%', #{name}, '%')
        </if>
        <if test="minAge != null">
            AND age >= #{minAge}
        </if>
        <if test="maxAge != null">
            AND age <= #{maxAge}
        </if>
        <if test="email != null and email != ''">
            AND email = #{email}
        </if>
        ORDER BY create_time DESC
    </select>
    <!-- 批量插入用戶 -->
    <insert id="batchInsert" parameterType="list">
        INSERT INTO user (name, email, age, create_time, update_time) 
        VALUES
        <foreach collection="list" item="user" separator=",">
            (#{user.name}, #{user.email}, #{user.age}, NOW(), NOW())
        </foreach>
    </insert>
    <!-- 統(tǒng)計(jì)用戶數(shù)量 -->
    <select id="countUsers" resultType="int">
        SELECT COUNT(*) FROM user WHERE deleted = 0
    </select>
</mapper>

@Service
@Slf4j
@Transactional
public class UserService {
    @Autowired
    private UserMapper userMapper;
    /**
     * 根據(jù)ID查詢用戶
     */
    public User getUserById(Long id) {
        log.info("查詢用戶: {}", id);
        User user = userMapper.selectById(id);
        if (user == null) {
            throw new RuntimeException("用戶不存在");
        }
        return user;
    }
    /**
     * 查詢所有用戶
     */
    public List<User> getAllUsers() {
        log.info("查詢所有用戶");
        return userMapper.selectAll();
    }
    /**
     * 創(chuàng)建用戶
     */
    public User createUser(User user) {
        log.info("創(chuàng)建用戶: {}", user.getName());
        // 檢查郵箱是否已存在
        User existingUser = userMapper.selectByEmail(user.getEmail());
        if (existingUser != null) {
            throw new RuntimeException("郵箱已存在");
        }
        int result = userMapper.insert(user);
        if (result > 0) {
            return user;
        } else {
            throw new RuntimeException("創(chuàng)建用戶失敗");
        }
    }
    /**
     * 更新用戶
     */
    public User updateUser(User user) {
        log.info("更新用戶: {}", user.getId());
        // 檢查用戶是否存在
        User existingUser = userMapper.selectById(user.getId());
        if (existingUser == null) {
            throw new RuntimeException("用戶不存在");
        }
        int result = userMapper.update(user);
        if (result > 0) {
            return getUserById(user.getId());
        } else {
            throw new RuntimeException("更新用戶失敗");
        }
    }
    /**
     * 刪除用戶（邏輯刪除）
     */
    public void deleteUser(Long id) {
        log.info("刪除用戶: {}", id);
        User existingUser = userMapper.selectById(id);
        if (existingUser == null) {
            throw new RuntimeException("用戶不存在");
        }
        int result = userMapper.deleteById(id);
        if (result == 0) {
            throw new RuntimeException("刪除用戶失敗");
        }
    }
    /**
     * 分頁查詢用戶
     */
    public PageResult<User> getUsersByPage(int pageNum, int pageSize) {
        log.info("分頁查詢用戶: 頁碼={}, 大小={}", pageNum, pageSize);
        int offset = (pageNum - 1) * pageSize;
        List<User> users = userMapper.selectByPage(offset, pageSize);
        int total = userMapper.countUsers();
        return new PageResult<>(users, total, pageNum, pageSize);
    }
    /**
     * 條件查詢用戶
     */
    public List<User> getUsersByCondition(String name, Integer minAge, Integer maxAge, String email) {
        log.info("條件查詢用戶: name={}, minAge={}, maxAge={}, email={}", 
                 name, minAge, maxAge, email);
        Map<String, Object> params = new HashMap<>();
        params.put("name", name);
        params.put("minAge", minAge);
        params.put("maxAge", maxAge);
        params.put("email", email);
        return userMapper.selectByCondition(params);
    }
    /**
     * 分頁結(jié)果封裝類
     */
    @Data
    @AllArgsConstructor
    public static class PageResult<T> {
        private List<T> data;
        private long total;
        private int pageNum;
        private int pageSize;
        private int totalPages;
        public PageResult(List<T> data, long total, int pageNum, int pageSize) {
            this.data = data;
            this.total = total;
            this.pageNum = pageNum;
            this.pageSize = pageSize;
            this.totalPages = (int) Math.ceil((double) total / pageSize);
        }
    }
}

@Configuration
@MapperScan("com.example.mapper")  // 掃描Mapper接口
@Slf4j
public class MyBatisConfig {
    /**
     * 配置MyBatis SqlSessionFactory
     */
    @Bean
    @ConfigurationProperties(prefix = "mybatis.configuration")
    public org.apache.ibatis.session.Configuration globalConfiguration() {
        org.apache.ibatis.session.Configuration configuration = 
            new org.apache.ibatis.session.Configuration();
        configuration.setMapUnderscoreToCamelCase(true);
        configuration.setLogImpl(StdOutImpl.class);
        configuration.setCacheEnabled(true);
        return configuration;
    }
    /**
     * 分頁插件配置
     */
    @Bean
    public MybatisPlusInterceptor mybatisPlusInterceptor() {
        MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
        // 分頁插件
        PaginationInnerInterceptor paginationInterceptor = new PaginationInnerInterceptor();
        paginationInterceptor.setMaxLimit(1000L);  // 單頁最大記錄數(shù)
        paginationInterceptor.setOverflow(true);   // 超過最大頁數(shù)時(shí)返回第一頁
        interceptor.addInnerInterceptor(paginationInterceptor);
        // 樂觀鎖插件（可選）
        OptimisticLockerInnerInterceptor optimisticLockerInterceptor = 
            new OptimisticLockerInnerInterceptor();
        interceptor.addInnerInterceptor(optimisticLockerInterceptor);
        log.info("MyBatis 插件配置完成");
        return interceptor;
    }
    /**
     * 元對(duì)象處理器（自動(dòng)填充字段）
     */
    @Bean
    public MetaObjectHandler metaObjectHandler() {
        return new MetaObjectHandler() {
            @Override
            public void insertFill(MetaObject metaObject) {
                this.strictInsertFill(metaObject, "createTime", LocalDateTime.class, LocalDateTime.now());
                this.strictInsertFill(metaObject, "updateTime", LocalDateTime.class, LocalDateTime.now());
            }
            @Override
            public void updateFill(MetaObject metaObject) {
                this.strictUpdateFill(metaObject, "updateTime", LocalDateTime.class, LocalDateTime.now());
            }
        };
    }
}

<!-- pom.xml -->
<dependencies>
    <!-- Kafka Starter -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- JSON 支持 -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
</dependencies>

# application.yml
spring:
  kafka:
    # Kafka 服務(wù)器配置
    bootstrap-servers: localhost:9092
    # 生產(chǎn)者配置
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all  # 所有副本確認(rèn)
      retries: 3  # 重試次數(shù)
      batch-size: 16384  # 批量大小
      buffer-memory: 33554432  # 緩沖區(qū)大小
      properties:
        linger.ms: 10  # 發(fā)送延遲
    # 消費(fèi)者配置
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      group-id: spring-boot-demo-group  # 消費(fèi)者組
      auto-offset-reset: earliest  # 偏移量重置策略
      enable-auto-commit: false  # 手動(dòng)提交偏移量
      properties:
        spring.json.trusted.packages: "*"  # 信任所有包進(jìn)行反序列化
    # 監(jiān)聽器配置
    listener:
      ack-mode: manual  # 手動(dòng)提交模式
      concurrency: 3    # 并發(fā)消費(fèi)者數(shù)量

@Configuration
@Slf4j
@EnableKafka
public class KafkaConfig {
    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;
    /**
     * 生產(chǎn)者工廠配置
     */
    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        return new DefaultKafkaProducerFactory<>(props);
    }
    /**
     * Kafka 模板
     */
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        KafkaTemplate<String, Object> template = new KafkaTemplate<>(producerFactory());
        template.setProducerListener(new ProducerListener<String, Object>() {
            @Override
            public void onSuccess(ProducerRecord<String, Object> record, RecordMetadata metadata) {
                log.info("消息發(fā)送成功: topic={}, partition={}, offset={}", 
                    record.topic(), metadata.partition(), metadata.offset());
            }
            @Override
            public void onError(ProducerRecord<String, Object> record, Exception exception) {
                log.error("消息發(fā)送失敗: topic={}, key={}", record.topic(), record.key(), exception);
            }
        });
        return template;
    }
    /**
     * 消費(fèi)者工廠配置
     */
    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "spring-boot-demo-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        return new DefaultKafkaConsumerFactory<>(props);
    }
    /**
     * 監(jiān)聽器容器工廠
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);  // 并發(fā)消費(fèi)者數(shù)量
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        // 錯(cuò)誤處理
        factory.setErrorHandler(((thrownException, data) -> {
            log.error("消費(fèi)消息失敗: {}", data, thrownException);
        }));
        return factory;
    }
}

@Service
@Slf4j
public class KafkaProducerService {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    /**
     * 發(fā)送用戶注冊(cè)消息
     */
    public void sendUserRegistration(User user) {
        String topic = "user-registration";
        String key = "user_" + user.getId();
        UserRegistrationEvent event = new UserRegistrationEvent(
            user.getId(), user.getName(), user.getEmail(), LocalDateTime.now());
        // 發(fā)送消息
        CompletableFuture<SendResult<String, Object>> future = 
            kafkaTemplate.send(topic, key, event);
        // 異步處理發(fā)送結(jié)果
        future.whenComplete((result, exception) -> {
            if (exception != null) {
                log.error("發(fā)送用戶注冊(cè)消息失敗: userId={}", user.getId(), exception);
            } else {
                log.info("用戶注冊(cè)消息發(fā)送成功: userId={}, offset={}", 
                    user.getId(), result.getRecordMetadata().offset());
            }
        });
    }
    /**
     * 發(fā)送訂單創(chuàng)建消息
     */
    public void sendOrderCreation(Order order) {
        String topic = "order-creation";
        String key = "order_" + order.getId();
        // 確保消息順序：使用訂單ID作為key確保同一訂單的消息發(fā)送到同一分區(qū)
        kafkaTemplate.send(topic, key, order)
            .addCallback(
                result -> log.info("訂單創(chuàng)建消息發(fā)送成功: orderId={}", order.getId()),
                exception -> log.error("訂單創(chuàng)建消息發(fā)送失敗: orderId={}", order.getId(), exception)
            );
    }
    /**
     * 發(fā)送帶事務(wù)的消息
     */
    @Transactional
    public void sendTransactionalMessage(String topic, String key, Object message) {
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send(topic, key, message);
            // 這里可以添加數(shù)據(jù)庫操作，確保消息和數(shù)據(jù)庫操作在同一個(gè)事務(wù)中
            return null;
        });
    }
    /**
     * 用戶注冊(cè)事件
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class UserRegistrationEvent {
        private Long userId;
        private String username;
        private String email;
        private LocalDateTime registerTime;
    }
    /**
     * 訂單實(shí)體
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Order {
        private Long id;
        private Long userId;
        private BigDecimal amount;
        private LocalDateTime createTime;
    }
}

@Service
@Slf4j
public class KafkaConsumerService {
    @Autowired
    private EmailService emailService;
    @Autowired
    private UserService userService;
    /**
     * 消費(fèi)用戶注冊(cè)消息
     */
    @KafkaListener(topics = "user-registration", groupId = "user-service")
    public void consumeUserRegistration(
            @Payload KafkaProducerService.UserRegistrationEvent event,
            @Header(KafkaHeaders.RECEIVED_KEY) String key,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            Acknowledgment ack) {
        log.info("收到用戶注冊(cè)消息: key={}, partition={}, event={}", key, partition, event);
        try {
            // 1. 發(fā)送歡迎郵件
            emailService.sendWelcomeEmail(event.getEmail(), event.getUsername());
            // 2. 初始化用戶積分
            userService.initUserPoints(event.getUserId());
            // 3. 記錄注冊(cè)日志
            log.info("用戶注冊(cè)處理完成: userId={}", event.getUserId());
            // 4. 手動(dòng)提交偏移量
            ack.acknowledge();
        } catch (Exception e) {
            log.error("處理用戶注冊(cè)消息失敗: userId={}", event.getUserId(), e);
            // 根據(jù)業(yè)務(wù)需求決定是否重試
        }
    }
    /**
     * 消費(fèi)訂單創(chuàng)建消息
     */
    @KafkaListener(topics = "order-creation", groupId = "order-service")
    public void consumeOrderCreation(
            @Payload KafkaProducerService.Order order,
            @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long timestamp,
            Acknowledgment ack) {
        log.info("收到訂單創(chuàng)建消息: orderId={}, amount={}, time={}", 
            order.getId(), order.getAmount(), 
            Instant.ofEpochMilli(timestamp).atZone(ZoneId.systemDefault()).toLocalDateTime());
        try {
            // 1. 庫存檢查
            checkInventory(order);
            // 2. 風(fēng)控檢查
            riskControlCheck(order);
            // 3. 發(fā)送訂單確認(rèn)
            sendOrderConfirmation(order);
            log.info("訂單處理完成: orderId={}", order.getId());
            ack.acknowledge();
        } catch (Exception e) {
            log.error("處理訂單消息失敗: orderId={}", order.getId(), e);
            // 根據(jù)異常類型決定處理策略
            if (e instanceof InventoryException) {
                // 庫存不足，需要人工處理
                handleInventoryShortage(order, e);
            } else {
                // 其他異常，可以重試
                throw e;
            }
        }
    }
    /**
     * 批量消費(fèi)消息
     */
    @KafkaListener(topics = "batch-processing", groupId = "batch-service")
    public void consumeBatchMessages(
            List<ConsumerRecord<String, Object>> records,
            Acknowledgment ack) {
        log.info("收到批量消息，數(shù)量: {}", records.size());
        try {
            for (ConsumerRecord<String, Object> record : records) {
                processSingleRecord(record);
            }
            // 批量提交偏移量
            ack.acknowledge();
        } catch (Exception e) {
            log.error("批量處理消息失敗", e);
            // 可以根據(jù)業(yè)務(wù)需求實(shí)現(xiàn)重試邏輯
        }
    }
    // 模擬業(yè)務(wù)方法
    private void checkInventory(KafkaProducerService.Order order) {
        // 庫存檢查邏輯
        log.info("檢查庫存: orderId={}", order.getId());
    }
    private void riskControlCheck(KafkaProducerService.Order order) {
        // 風(fēng)控檢查邏輯
        log.info("風(fēng)控檢查: orderId={}", order.getId());
    }
    private void sendOrderConfirmation(KafkaProducerService.Order order) {
        // 發(fā)送確認(rèn)邏輯
        log.info("發(fā)送訂單確認(rèn): orderId={}", order.getId());
    }
    private void handleInventoryShortage(KafkaProducerService.Order order, Exception e) {
        // 處理庫存不足
        log.warn("處理庫存不足: orderId={}", order.getId());
    }
    private void processSingleRecord(ConsumerRecord<String, Object> record) {
        // 處理單條記錄
        log.info("處理記錄: key={}, value={}", record.key(), record.value());
    }
}

@Service
@Slf4j
public class EmailService {
    /**
     * 發(fā)送歡迎郵件
     */
    public void sendWelcomeEmail(String email, String username) {
        log.info("發(fā)送歡迎郵件: 收件人={}, 用戶名={}", email, username);
        // 模擬郵件發(fā)送
        try {
            Thread.sleep(100); // 模擬網(wǎng)絡(luò)延遲
            log.info("歡迎郵件發(fā)送成功: {}", email);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("郵件發(fā)送中斷", e);
        }
    }
    /**
     * 發(fā)送訂單確認(rèn)郵件
     */
    public void sendOrderConfirmation(String email, String orderInfo) {
        log.info("發(fā)送訂單確認(rèn)郵件: 收件人={}, 訂單信息={}", email, orderInfo);
        // 模擬郵件發(fā)送
        try {
            Thread.sleep(150);
            log.info("訂單確認(rèn)郵件發(fā)送成功: {}", email);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("郵件發(fā)送中斷", e);
        }
    }
}

@Component
public class KafkaHealthIndicator implements HealthIndicator {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    @Override
    public Health health() {
        try {
            // 測(cè)試Kafka連接
            Future<RecordMetadata> future = kafkaTemplate.send("health-check", "test-key", "test-value");
            // 設(shè)置超時(shí)時(shí)間
            try {
                RecordMetadata metadata = future.get(5, TimeUnit.SECONDS);
                return Health.up()
                    .withDetail("cluster", "kafka")
                    .withDetail("topic", metadata.topic())
                    .withDetail("partition", metadata.partition())
                    .withDetail("status", "connected")
                    .build();
            } catch (TimeoutException e) {
                return Health.down()
                    .withDetail("error", "Kafka connection timeout")
                    .build();
            }
        } catch (Exception e) {
            return Health.down(e)
                .withDetail("error", "Kafka health check failed: " + e.getMessage())
                .build();
        }
    }
}

# 生產(chǎn)環(huán)境 Redis 優(yōu)化配置
spring:
  redis:
    lettuce:
      pool:
        max-active: 50
        max-idle: 20
        min-idle: 10
    timeout: 1000ms

// 啟用二級(jí)緩存
@CacheNamespace
public interface UserMapper {
    // Mapper 接口
}
// 使用批量操作
@Insert("<script>INSERT INTO user (...) VALUES " +
       "<foreach collection='list' item='item' separator=','>(...)</foreach></script>")
void batchInsert(List<User> users);

spring:
  kafka:
    producer:
      batch-size: 32768      # 增大批量大小
      linger-ms: 20          # 適當(dāng)增加延遲
      compression-type: snappy # 啟用壓縮
    consumer:
      fetch-max-wait: 500    # 最大等待時(shí)間
      fetch-min-size: 1024   # 最小獲取大小