#削峰填谷

// 内存队列：简单但有风险
public class MemoryQueueProcessor {
    private BlockingQueue<Request> queue = new LinkedBlockingQueue<>(10000);

    public boolean submit(Request req) {
        // ✅ 队列有容量，返回 false 表示拒绝
        // ❌ 队列满了，请求被丢弃
        return queue.offer(req, 3, TimeUnit.SECONDS);
    }

    @Async
    public void consume() {
        while (true) {
            Request req = queue.poll(1, TimeUnit.SECONDS);
            if (req != null) {
                process(req);
            }
        }
    }
}

// MQ 队列：分布式削峰
public class MQProcessor {
    private KafkaTemplate<String, Request> kafkaTemplate;

    public boolean submit(Request req) {
        // Kafka 异步发送，不阻塞
        kafkaTemplate.send("activity-topic", req.getKey(), req);
        return true; // 发送成功即返回
    }

    // 消费者按自己的节奏处理
    @KafkaListener(topics = "activity-topic", groupId = "activity-consumer")
    public void consume(Request req) {
        process(req);
    }
}

// 批量写入：每 100 条或每 500ms 写一次
public class BatchWriter {
    private BlockingQueue<Order> buffer = new LinkedBlockingQueue<>(10000);
    private ScheduledExecutorService flusher = Executors.newSingleThreadScheduledExecutor();

    public void submit(Order order) {
        buffer.offer(order);
    }

    public void start() {
        // 每 500ms 刷新一次
        flusher.scheduleAtFixedRate(() -> {
            List<Order> batch = new ArrayList<>();
            buffer.drainTo(batch, 100);  // 最多取 100 条

            if (!batch.isEmpty()) {
                orderMapper.batchInsert(batch);
                logger.info("批量写入 {} 条订单", batch.size());
            }
        }, 0, 500, TimeUnit.MILLISECONDS);
    }
}

public class DeduplicatedReader<T> {
    private LoadingCache<String, T> localCache;
    private ConcurrentHashMap<String, CountDownLatch> pending = new ConcurrentHashMap<>();

    public T get(String key, Supplier<T> loader) {
        // 第一步：检查本地缓存
        T cached = localCache.getIfPresent(key);
        if (cached != null) return cached;

        // 第二步：检查是否已有其他请求在加载
        CountDownLatch latch = pending.computeIfAbsent(key, k -> new CountDownLatch(1));

        if (latch.getCount() == 0) {
            // 已有请求完成，重新查缓存
            return localCache.get(key);
        }

        // 第三步：自己加载
        try {
            T value = loader.get();
            localCache.put(key, value);
            return value;
        } finally {
            latch.countDown();
            pending.remove(key);
        }
    }
}

public class ThrottledController {
    private Semaphore permits = new Semaphore(5000); // 同时处理 5000 个请求
    private AtomicLong queueWaitEstimate = new AtomicLong(0);

    @GetMapping("/seckill")
    public Result<SeckillToken> seckill(Long userId, Long skuId) {
        // 尝试获取处理许可
        if (!permits.tryAcquire()) {
            long waitTime = queueWaitEstimate.get();
            if (waitTime > 30000) {
                // 排队超过 30 秒，建议用户放弃
                return Result.fail("活动过于火爆，您来晚了，明天再来试试？");
            } else {
                // 还在排队，给个进度提示
                return Result.fail("当前排队人数较多，预计等待 " + (waitTime / 1000) + " 秒");
            }
        }

        try {
            return doSeckill(userId, skuId);
        } finally {
            permits.release();
        }
    }
}

// ❌ 错误：消费逻辑中没有超时控制
@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "order-consumer"
)
public class OrderConsumer implements RocketMQListener<OrderMessage> {
    @Override
    public void onMessage(OrderMessage message) {
        // 数据库操作没有超时控制
        // 如果数据库慢，这个线程会永久阻塞
        orderService.process(message); // 可能在 DB 耗时 30 秒
    }
}

// ✅ 正确：消费逻辑加超时控制 + 独立线程池
@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "order-consumer",
    consumeThreadCount = 50,  // 消费线程数
    consumeTimeout = 15       // 消费超时 15 秒
)
public class OrderConsumer implements RocketMQListener<OrderMessage> {
    private ThreadPoolExecutor executor = new ThreadPoolExecutor(
        20, 50, 60, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(1000),
        new ThreadFactoryBuilder().setNameFormat("order-consume-%d").build(),
        new ThreadPoolExecutor.CallerRunsPolicy()  // 队列满时由调用线程执行
    );

    @Override
    public void onMessage(OrderMessage message) {
        executor.submit(() -> {
            try {
                // 加超时控制
                CompletableFuture.runAsync(() -> orderService.process(message))
                    .get(10, TimeUnit.SECONDS);
            } catch (TimeoutException e) {
                logger.warn("消费超时 messageId={}", message.getId());
                // 放回队列重试，或者记录死信
            }
        });
    }
}