#消息积压处理策略

// 如果是Kafka，扩容Consumer实例
// 注意：Consumer数不能超过Partition数
// 如果超过了，多余的Consumer会空转

// 查看当前分区数
kafka-topics.sh --describe --topic order-events --bootstrap-server kafka:9092

// 如果分区数不够，先增加分区
kafka-topics.sh --alter --topic order-events \
  --partitions 16 --bootstrap-server kafka:9092

// 生产端限流：降低生产速度，为消费端争取时间

// 方案1：降低发送频率
Thread.sleep(100);  // 每条消息间隔100ms

// 方案2：使用令牌桶限流
RateLimiter limiter = RateLimiter.create(1000);  // 每秒1000条
for (Message msg : messages) {
    limiter.acquire();
    producer.send(msg);
}

// 方案3：直接暂停生产（极端情况）
producer.pause();  // 暂停发送
// 处理完积压后
producer.resume();  // 恢复发送

// 积压期间，临时优化Consumer配置

// 1. 增加每次拉取的消息数量
props.put("max.poll.records", "2000");  // 默认500条

// 2. 减少poll间隔
props.put("fetch.min.bytes", "1");  // 有消息就拉

// 3. 关闭部分校验逻辑（积压期间降级）
// 积压处理优先，不做详细的业务校验

// 4. 使用批量处理
List<Message> batch = new ArrayList<>();
while ((msg = consumer.poll(100)) != null) {
    batch.add(msg);
    if (batch.size() >= 100) {
        processBatch(batch);  // 批量处理
        batch.clear();
    }
}

// 方案：优先消费"有价值"的消息，跳过"无效"消息

while (true) {
    List<MessageExt> msgs = consumer.poll(1000);

    // 按消息年龄分组
    Map<Long, List<MessageExt>> grouped =
        msgs.stream().collect(Collectors.groupingBy(msg -> {
            return (System.currentTimeMillis() - msg.getBornTimestamp()) / 60000;
        }));

    // 优先处理最近的消息（可能还在时效内）
    grouped.keySet().stream().sorted()
        .forEach(age -> {
            for (MessageExt msg : grouped.get(age)) {
                if (isStillValid(msg)) {
                    processMessage(msg);
                } else {
                    skipMessage(msg);  // 跳过无效消息
                }
            }
        });
}

// 1. 设置重试上限
int maxRetry = 3;
int retryCount = getRetryCount(msg.getMsgId());  // 从Redis/DB获取重试次数

if (retryCount >= maxRetry) {
    // 超过重试次数，投递到死信队列（DLQ）
    sendToDeadLetterQueue(msg);
    return ConsumeStatus.SUCCESS;  // 返回成功，不再重试
}

// 2. 区分可重试和不可重试的异常
try {
    processMessage(msg);
} catch (TransientException e) {
    // 临时异常（网络、数据库），可以重试
    return ConsumeStatus.RECONSUME_LATER;
} catch (FatalException e) {
    // 致命异常（数据格式、业务错误），直接投递DLQ
    sendToDeadLetterQueue(msg);
    return ConsumeStatus.SUCCESS;
}

// 1. 幂等消费：业务层去重
public void processMessage(Message msg) {
    String msgId = msg.getMsgId();
    String key = "consumed:msg:" + msgId;

    // Redis SETNX 去重
    boolean alreadyConsumed = redis.setIfAbsent(key, "1",
        Duration.ofDays(7));

    if (alreadyConsumed) {
        doProcess(msg);
    } else {
        log.info("消息已处理过，跳过: {}", msgId);
    }
}

// 2. 扩容时避免Rebalance
// 使用StickyAssignor
props.put("partition.assignment.strategy",
    "org.apache.kafka.clients.consumer.StickyAssignor");

// 或者：扩容时先暂停消费，处理完积压再扩容