深度解析：分布式系统各组件

// 1. 轮询（Round Robin）- 最简单
请求1 → 服务器1
请求2 → 服务器2
请求3 → 服务器3
请求4 → 服务器1  // 循环
请求5 → 服务器2

// 2. 加权轮询（Weighted Round Robin）
服务器1：权重3（性能好）
服务器2：权重2（性能中等）
服务器3：权重1（性能差）

10个请求的分配：
服务器1处理5个
服务器2处理3个
服务器3处理2个

// 3. 最少连接（Least Connections）
服务器1：当前10个连接
服务器2：当前5个连接  ← 新请求给它
服务器3：当前8个连接

// 4. IP哈希（IP Hash）
用户IP:  192.168.1.100 → hash → 永远分配到服务器2
好处：同一用户总是访问同一台服务器（会话保持）

// Spring Boot配置Redis Session
@Configuration
@EnableRedisHttpSession(maxInactiveIntervalInSeconds = 1800) // 30分钟
public class SessionConfig {
    // 自动配置，Session会存到Redis
}

// 使用时完全透明
@RestController
public class UserController {
    
    @PostMapping("/login")
    public Result login(HttpSession session, String username, String password) {
        // 验证用户名密码... 
        User user = userService.login(username, password);
        
        // 存入Session（自动存到Redis）
        session.setAttribute("user", user);
        
        return Result.success();
    }
    
    @GetMapping("/profile")
    public Result getProfile(HttpSession session) {
        // 从Session获取（自动从Redis读）
        User user = (User) session.getAttribute("user");
        
        if (user == null) {
            return Result. error("未登录");
        }
        
        return Result.success(user);
    }
}

// 1. 配置Kafka生产者
Properties props = new Properties();
props.put("bootstrap. servers", "192.168.1.100:9092");  // Kafka地址
props.put("key.serializer", "org.apache.kafka.common.serialization. StringSerializer");
props.put("value.serializer", "org. apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 2. 创建订单后发送消息
public void createOrder(Order order) {
    // 保存订单到MySQL
    orderDao.save(order);
    
    // 发送消息到Kafka
    ProducerRecord<String, String> record = new ProducerRecord<>(
        "order-topic",              // Topic名称
        order.getUserId(),          // Key（用于分区）
        JSON.toJSONString(order)    // Value（消息内容）
    );
    
    // 异步发送
    producer.send(record, new Callback() {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception e) {
            if (e != null) {
                log.error("发送失败", e);
                // 可以重试或记录到数据库
            } else {
                log.info("发送成功，Partition: {}, Offset: {}", 
                    metadata.partition(), metadata.offset());
            }
        }
    });
}

// 1. 配置Kafka消费者
Properties props = new Properties();
props.put("bootstrap.servers", "192.168.1.100:9092");
props.put("group.id", "inventory-service");  // 消费者组ID
props.put("key.deserializer", "org.apache. kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 2. 订阅Topic
consumer.subscribe(Arrays.asList("order-topic"));

// 3. 持续拉取消息
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        // 解析消息
        Order order = JSON.parseObject(record.value(), Order.class);
        
        // 处理业务逻辑
        inventoryService.reduceStock(order. getProductId(), order.getQuantity());
        
        System.out.printf("Partition: %d, Offset: %d, Key: %s, Value: %s%n",
            record.partition(), record.offset(), record.key(), record.value());
    }
    
    // 手动提交offset（确保消息处理成功后再提交）
    consumer. commitSync();
}

// 策略1：根据Key的Hash分区（相同Key进同一分区，保证顺序）
record = new ProducerRecord<>("order-topic", userId, orderData);
// userId相同的订单会进入同一个分区，保证该用户的订单顺序

// 策略2：轮询分区（负载均衡）
record = new ProducerRecord<>("order-topic", null, orderData);
// Key为null时，轮流发送到各个分区

// 策略3：自定义分区器
public class CustomPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                        Object value, byte[] valueBytes, Cluster cluster) {
        // VIP用户发到分区0（高优先级处理）
        if (isVipUser(key)) {
            return 0;
        }
        // 普通用户发到其他分区
        return Utils.toPositive(Utils.murmur2(keyBytes)) % (cluster.partitionCountForTopic(topic) - 1) + 1;
    }
}

// 1. 自动提交（简单但可能丢消息）
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");  // 每秒自动提交

// 风险：消息取出后还没处理完，程序崩溃，offset已提交 → 消息丢失

// 2. 手动同步提交（安全但慢）
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);  // 处理消息
    }
    consumer.commitSync();  // 处理完才提交，阻塞等待确认
}

// 3. 手动异步提交（性能好）
consumer.commitAsync(new OffsetCommitCallback() {
    @Override
    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
        if (e != null) {
            log.error("提交失败", e);
        }
    }
});

// 用户点击秒杀按钮
@PostMapping("/seckill")
public Result seckill(Long productId, Long userId) {
    // 1. 立即返回"排队中"
    String requestId = UUID.randomUUID().toString();
    
    // 2. 发送到Kafka（异步处理）
    SeckillRequest request = new SeckillRequest(requestId, productId, userId);
    kafkaTemplate.send("seckill-topic", request);
    
    // 3. 立即返回给用户
    return Result.success("您的请求正在处理，请稍候查询结果", requestId);
}

// 消费者慢慢处理
@KafkaListener(topics = "seckill-topic", concurrency = "10")  // 10个线程并行
public void handleSeckill(SeckillRequest request) {
    // 1. 检查库存（Redis）
    Long stock = redisTemplate.opsForValue().decrement("stock:" + request.getProductId());
    
    if (stock >= 0) {
        // 2. 创建订单（MySQL）
        Order order = createOrder(request);
        
        // 3. 通知用户成功
        notifyUser(request. getUserId(), "秒杀成功！");
    } else {
        // 4. 恢复库存
        redisTemplate.opsForValue().increment("stock:" + request.getProductId());
        
        // 5. 通知用户失败
        notifyUser(request.getUserId(), "商品已售罄");
    }
}

// 各个服务打日志
public class KafkaLogAppender extends AppenderBase<ILoggingEvent> {
    @Override
    protected void append(ILoggingEvent event) {
        String log = event.getFormattedMessage();
        kafkaProducer.send(new ProducerRecord<>("app-logs", log));
    }
}

// 日志消费者（ELK架构）
@KafkaListener(topics = "app-logs")
public void consumeLog(String log) {
    // 解析日志
    LogEntry entry = parseLog(log);
    
    // 存入Elasticsearch（方便搜索）
    elasticsearchTemplate.save(entry);
    
    // 如果是ERROR级别，发送告警
    if (entry.getLevel().equals("ERROR")) {
        alertService.sendAlert(entry);
    }
}

// MySQL数据变化 → 同步到ES
@KafkaListener(topics = "mysql-binlog")
public void syncToElasticsearch(BinlogEvent event) {
    if (event.getType() == EventType.INSERT) {
        // 新增数据
        elasticsearchRepository.save(event.getData());
    } else if (event.getType() == EventType.UPDATE) {
        // 更新数据
        elasticsearchRepository.update(event.getData());
    } else if (event.getType() == EventType.DELETE) {
        // 删除数据
        elasticsearchRepository.delete(event.getId());
    }
}

// 页面访问计数
public void recordPageView(String pageId) {
    redisTemplate.opsForValue().increment("page: view:" + pageId);
}

// 获取访问量
public Long getPageView(String pageId) {
    return redisTemplate.opsForValue().get("page:view:" + pageId);
}

// 分布式锁
public boolean tryLock(String key, int expireSeconds) {
    return redisTemplate. opsForValue()
        .setIfAbsent(key, "locked", expireSeconds, TimeUnit.SECONDS);
}

// 存储用户对象
public void saveUser(User user) {
    Map<String, String> userMap = new HashMap<>();
    userMap.put("name", user. getName());
    userMap.put("age", String.valueOf(user.getAge()));
    userMap.put("city", user.getCity());
    
    redisTemplate.opsForHash().putAll("user:" + user.getId(), userMap);
}

// 获取用户对象
public User getUser(Long userId) {
    Map<Object, Object> entries = redisTemplate.opsForHash()
        .entries("user:" + userId);
    
    User user = new User();
    user.setName((String) entries.get("name"));
    user.setAge(Integer.parseInt((String) entries.get("age")));
    user.setCity((String) entries.get("city"));
    return user;
}

// 购物车加商品
public void addToCart(Long userId, Long productId, int quantity) {
    redisTemplate.opsForHash()
        .increment("cart: user:" + userId, "product:" + productId, quantity);
}

// 发布文章（加入最新列表）
public void publishArticle(Long articleId) {
    redisTemplate.opsForList().leftPush("articles: latest", articleId. toString());
    
    // 保持列表只有100条
    redisTemplate. opsForList().trim("articles:latest", 0, 99);
}

// 获取最新文章
public List<String> getLatestArticles(int count) {
    return redisTemplate.opsForList().range("articles:latest", 0, count - 1);
}

// 简单消息队列
public void sendTask(String task) {
    redisTemplate. opsForList().leftPush("task:queue", task);
}

public String getTask() {
    return redisTemplate. opsForList().rightPop("task:queue");
}

// 用户点赞
public void likeArticle(Long articleId, Long userId) {
    redisTemplate.opsForSet().add("article:" + articleId + ":likes", userId. toString());
}

// 取消点赞
public void unlikeArticle(Long articleId, Long userId) {
    redisTemplate.opsForSet().remove("article:" + articleId + ":likes", userId.toString());
}

// 是否已点赞
public boolean hasLiked(Long articleId, Long userId) {
    return redisTemplate.opsForSet()
        .isMember("article:" + articleId + ":likes", userId.toString());
}

// 点赞总数
public Long getLikeCount(Long articleId) {
    return redisTemplate.opsForSet().size("article:" + articleId + ": likes");
}

// 共同好友
public Set<String> getCommonFriends(Long userId1, Long userId2) {
    return redisTemplate.opsForSet()
        .intersect("user:" + userId1 + ":friends", "user:" + userId2 + ":friends");
}

// 添加用户分数
public void addScore(Long userId, double score) {
    redisTemplate.opsForZSet().add("rank:score", userId.toString(), score);
}

// 增加分数
public void incrementScore(Long userId, double delta) {
    redisTemplate. opsForZSet().incrementScore("rank:score", userId. toString(), delta);
}

// 获取排行榜（前N名）
public Set<ZSetOperations.TypedTuple<String>> getTopN(int n) {
    return redisTemplate.opsForZSet()
        .reverseRangeWithScores("rank:score", 0, n - 1);
}

// 获取用户排名
public Long getUserRank(Long userId) {
    return redisTemplate.opsForZSet()
        .reverseRank("rank:score", userId.toString());
}

// 延时任务队列
public void addDelayTask(String taskId, long executeTime) {
    redisTemplate.opsForZSet().add("delay:queue", taskId, executeTime);
}

// 获取到期任务
public Set<String> getExpiredTasks() {
    long now = System.currentTimeMillis();
    return redisTemplate. opsForZSet()
        .rangeByScore("delay:queue", 0, now);
}

// 读数据
public Product getProduct(Long id) {
    // 1. 先查缓存
    String cacheKey = "product:" + id;
    Product product = redisTemplate.opsForValue().get(cacheKey);
    
    if (product != null) {
        return product;  // 缓存命中
    }
    
    // 2. 缓存未命中，查数据库
    product = productDao.selectById(id);
    
    if (product != null) {
        // 3. 写入缓存
        redisTemplate.opsForValue().set(cacheKey, product, 1, TimeUnit.HOURS);
    }
    
    return product;
}

// 写数据
public void updateProduct(Product product) {
    // 1. 先更新数据库
    productDao.updateById(product);
    
    // 2. 删除缓存（而不是更新缓存）
    redisTemplate.delete("product:" + product. getId());
    
    // 为什么删除而不是更新？
    // 因为可能有多个请求同时更新，删除缓存更安全
}

// 使用Spring Cache自动管理
@Cacheable(value = "products", key = "#id")
public Product getProduct(Long id) {
    // Spring自动处理：先查缓存，未命中则执行方法并缓存结果
    return productDao.selectById(id);
}

@CachePut(value = "products", key = "#product.id")
public Product updateProduct(Product product) {
    // Spring自动处理：更新数据库后更新缓存
    productDao.updateById(product);
    return product;
}

@CacheEvict(value = "products", key = "#id")
public void deleteProduct(Long id) {
    // Spring自动处理：删除数据库后删除缓存
    productDao.deleteById(id);
}

// 方案1：缓存空值
public Product getProduct(Long id) {
    String cacheKey = "product:" + id;
    Product product = redisTemplate.opsForValue().get(cacheKey);
    
    if (product != null) {
        if (product.getId() == null) {
            return null;  // 之前查过，数据库没有
        }
        return product;
    }
    
    product = productDao.selectById(id);
    
    if (product == null) {
        // 缓存一个空对象，过期时间短一点
        redisTemplate. opsForValue().set(cacheKey, new Product(), 5, TimeUnit.MINUTES);
    } else {
        redisTemplate. opsForValue().set(cacheKey, product, 1, TimeUnit.HOURS);
    }
    
    return product;
}

// 方案2：布隆过滤器（Bloom Filter）
@Autowired
private RedissonClient redissonClient;

public Product getProduct(Long id) {
    // 1. 布隆过滤器判断是否存在
    RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter("product:bloom");
    if (! bloomFilter.contains(id)) {
        return null;  // 一定不存在，直接返回
    }
    
    // 2. 可能存在，查缓存和数据库
    // ... 正常逻辑
}

// 初始化布隆过滤器
public void initBloomFilter() {
    RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter("product:bloom");
    bloomFilter.tryInit(100000, 0.01);  // 预计10万条数据，1%误判率
    
    // 加载所有商品ID
    List<Long> productIds = productDao.selectAllIds();
    productIds.forEach(bloomFilter::add);
}

// 方案1：互斥锁（只让一个请求查数据库）
public Product getProduct(Long id) {
    String cacheKey = "product:" + id;
    Product product = redisTemplate.opsForValue().get(cacheKey);
    
    if (product != null) {
        return product;
    }
    
    // 尝试获取锁
    String lockKey = "lock:product:" + id;
    boolean locked = redisTemplate.opsForValue()
        .setIfAbsent(lockKey, "1", 10, TimeUnit. SECONDS);
    
    if (locked) {
        try {
            // 拿到锁，查数据库
            product = productDao.selectById(id);
            redisTemplate. opsForValue().set(cacheKey, product, 1, TimeUnit.HOURS);
            return product;
        } finally {
            redisTemplate. delete(lockKey);  // 释放锁
        }
    } else {
        // 没拿到锁，等待一下再查缓存
        Thread.sleep(50);
        return getProduct(id);  // 递归调用
    }
}

// 方案2：永不过期（逻辑过期）
public Product getProduct(Long id) {
    String cacheKey = "product:" + id;
    CacheData cacheData = redisTemplate. opsForValue().get(cacheKey);
    
    if (cacheData == null) {
        // 缓存重建
        return rebuildCache(id);
    }
    
    // 检查逻辑过期时间
    if (cacheData.getExpireTime().isBefore(LocalDateTime.now())) {
        // 已过期，异步重建缓存
        executorService.submit(() -> rebuildCache(id));
        
        // 返回旧数据（虽然过期但可用）
        return cacheData. getProduct();
    }
    
    return cacheData.getProduct();
}

private Product rebuildCache(Long id) {
    // 获取互斥锁
    String lockKey = "lock:product:" + id;
    if (! tryLock(lockKey)) {
        return null;  // 有其他线程在重建
    }
    
    try {
        Product product = productDao.selectById(id);
        CacheData cacheData = new CacheData();
        cacheData.setProduct(product);
        cacheData.setExpireTime(LocalDateTime.now().plusHours(1));  // 逻辑过期时间
        
        redisTemplate.opsForValue().set("product:" + id, cacheData);  // 不设置Redis过期时间
        return product;
    } finally {
        unlock(lockKey);
    }
}

// 方案1：过期时间加随机值
public void cacheProduct(Product product) {
    int baseExpire = 3600;  // 1小时
    int randomExpire = new Random().nextInt(300);  // 0-5分钟随机
    int expire = baseExpire + randomExpire;
    
    redisTemplate.opsForValue().set(
        "product:" + product. getId(), 
        product, 
        expire, 
        TimeUnit.SECONDS
    );
}

// 方案2：热点数据永不过期
public void cacheHotProduct(Product product) {
    // 热门商品不设置过期时间
    redisTemplate.opsForValue().set("hot: product:" + product.getId(), product);
    
    // 定时任务定期更新
}

// 方案3：使用Redis集群 + 多级缓存
public Product getProduct(Long id) {
    // L1: 本地缓存（Caffeine）
    Product product = localCache.get(id);
    if (product != null) return product;
    
    // L2: Redis缓存
    product = redisTemplate. opsForValue().get("product:" + id);
    if (product != null) {
        localCache.put(id, product);
        return product;
    }
    
    // L3: 数据库
    product = productDao.selectById(id);
    if (product != null) {
        redisTemplate.opsForValue().set("product:" + id, product, 1, TimeUnit. HOURS);
        localCache.put(id, product);
    }
    
    return product;
}

@Autowired
private RedissonClient redissonClient;

// 简单使用
public void secKill(Long productId) {
    RLock lock = redissonClient.getLock("seckill:" + productId);
    
    try {
        // 尝试加锁，最多等待10秒，锁30秒后自动释放
        boolean locked = lock. tryLock(10, 30, TimeUnit.SECONDS);
        
        if (! locked) {
            throw new BusinessException("系统繁忙，请稍后重试");
        }
        
        // 执行业务逻辑
        int stock = getStock(productId);
        if (stock > 0) {
            reduceStock(productId);
            createOrder();
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } finally {
        lock.unlock();  // 释放锁
    }
}

// 可重入锁演示
public void demo() {
    RLock lock = redissonClient.getLock("mylock");
    lock.lock();
    
    try {
        method1();  // method1内部也尝试获取同一把锁
    } finally {
        lock.unlock();
    }
}

public void method1() {
    RLock lock = redissonClient.getLock("mylock");
    lock.lock();  // 可重入，不会死锁
    try {
        // 业务逻辑
    } finally {
        lock.unlock();
    }
}

// 联锁（MultiLock）
public void multiLock() {
    RLock lock1 = redissonClient.getLock("lock1");
    RLock lock2 = redissonClient.getLock("lock2");
    RLock lock3 = redissonClient.getLock("lock3");
    
    RedissonMultiLock multiLock = new RedissonMultiLock(lock1, lock2, lock3);
    
    try {
        multiLock. lock();
        // 同时持有3把锁
    } finally {
        multiLock.unlock();
    }
}

// 红锁（RedLock）- 多个独立Redis实例
public void redLock() {
    RLock lock1 = redissonClient1.getLock("lock");
    RLock lock2 = redissonClient2.getLock("lock");
    RLock lock3 = redissonClient3.getLock("lock");
    
    RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
    
    try {
        // 至少在N/2+1个实例上加锁成功才算成功
        redLock.lock();
    } finally {
        redLock.unlock();
    }
}

// 转账示例
@Transactional(rollbackFor = Exception.class)
public void transfer(Long fromUserId, Long toUserId, BigDecimal amount) {
    // 1. 扣减转出账户
    accountDao.deductBalance(fromUserId, amount);
    
    // 2. 模拟异常
    if (amount.compareTo(new BigDecimal("1000")) > 0) {
        throw new RuntimeException("单笔转账不能超过1000元");
    }
    
    // 3. 增加转入账户
    accountDao.addBalance(toUserId, amount);
    
    // 4. 记录流水
    transactionLogDao.insert(new TransactionLog(fromUserId, toUserId, amount));
}
// 如果抛出异常，1、3、4步骤都会回滚

// Spring Boot配置
spring: 
  data:
    mongodb:
      uri: mongodb://mongo1:27017,mongo2:27017,mongo3:27017/mydb? replicaSet=myReplicaSet&readPreference=secondaryPreferred

// 代码示例
@Autowired
private MongoTemplate mongoTemplate;

// 写操作（自动发送到Primary）
public void saveProduct(Product product) {
    mongoTemplate.save(product);
}

// 读操作（可以从Secondary读取，减轻Primary压力）
@ReadPreference(ReadPreferenceType.SECONDARY_PREFERRED)
public List<Product> findProducts(String category) {
    Query query = new Query(Criteria. where("category").is(category));
    return mongoTemplate.find(query, Product.class);
}

// 1. 添加依赖
<dependency>
    <groupId>com.ctrip.framework.apollo</groupId>
    <artifactId>apollo-client</artifactId>
    <version>2.0.1</version>
</dependency>

// 2. 配置文件
# application.properties
app.id=my-service
apollo.meta=http://apollo-config-server:8080
apollo.bootstrap. enabled=true
apollo.bootstrap.namespaces=application,database,redis

// 3. 启动类
@SpringBootApplication
@EnableApolloConfig
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

// 4. 使用配置
@Service
public class UserService {
    // 方式1：@Value注解
    @Value("${server.port:8080}")
    private int port;
    
    // 方式2：@ConfigurationProperties
    @Autowired
    private DataSourceProperties dataSourceProperties;
    
    // 方式3：监听配置变化
    @ApolloConfigChangeListener
    public void onChange(ConfigChangeEvent changeEvent) {
        for (String key : changeEvent.changedKeys()) {
            ConfigChange change = changeEvent.getChange(key);
            log.info("配置变化: {} - 旧值: {}, 新值: {}",
                key, change.getOldValue(), change.getNewValue());
            
            // 根据配置变化执行业务逻辑
            if ("max. pool.size".equals(key)) {
                updateThreadPool(Integer.parseInt(change.getNewValue()));
            }
        }
    }
}

@ConfigurationProperties(prefix = "spring.datasource")
@Data
public class DataSourceProperties {
    private String url;
    private String username;
    private String password;
}

@Component
public class CustomMetrics {
    private final MeterRegistry meterRegistry;
    
    public CustomMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }
    
    // 计数器（累计值）
    public void recordOrder() {
        meterRegistry.counter("order.created", "status", "success").increment();
    }
    
    // 计时器（统计耗时）
    public void recordApiDuration(String api, long duration) {
        meterRegistry.timer("api.duration", "path", api)
            .record(duration, TimeUnit.MILLISECONDS);
    }
    
    // 仪表盘（当前值）
    @PostConstruct
    public void init() {
        meterRegistry.gauge("thread.pool.active", threadPoolExecutor, ThreadPoolExecutor::getActiveCount);
    }
}

// 使用示例
@Service
public class OrderService {
    @Autowired
    private CustomMetrics customMetrics;
    
    public void createOrder(Order order) {
        long start = System.currentTimeMillis();
        
        try {
            // 业务逻辑
            orderDao.save(order);
            
            // 记录成功
            customMetrics.recordOrder();
        } finally {
            // 记录耗时
            customMetrics.recordApiDuration("/api/order/create", 
                System.currentTimeMillis() - start);
        }
    }
}

// 自定义健康检查
@Component
public class CustomHealthIndicator implements HealthIndicator {
    @Autowired
    private RedisTemplate redisTemplate;
    
    @Autowired
    private KafkaTemplate kafkaTemplate;
    
    @Override
    public Health health() {
        // 检查Redis连接
        try {
            redisTemplate.opsForValue().get("health-check");
        } catch (Exception e) {
            return Health.down()
                .withDetail("redis", "连接失败:  " + e.getMessage())
                .build();
        }
        
        // 检查Kafka连接
        // ... 
        
        // 检查数据库连接池
        HikariDataSource dataSource = getDataSource();
        int activeConnections = dataSource.getHikariPoolMXBean().getActiveConnections();
        int maxConnections = dataSource.getMaximumPoolSize();
        
        if (activeConnections > maxConnections * 0.9) {
            return Health.down()
                .withDetail("database", "连接池即将耗尽")
                .withDetail("active", activeConnections)
                .withDetail("max", maxConnections)
                .build();
        }
        
        return Health. up()
            .withDetail("redis", "正常")
            .withDetail("kafka", "正常")
            .withDetail("database", "正常")
            .build();
    }
}