/**
 * 1. Direct Exchange — 精确匹配路由键
 */
channel.exchangeDeclare("direct_exchange", BuiltinExchangeType.DIRECT);
channel.queueBind("order_queue", "direct_exchange", "order.key");
channel.queueBind("payment_queue", "direct_exchange", "payment.key");
// 路由键 = "order.key" → order_queue
// 路由键 = "payment.key" → payment_queue

/**
 * 2. Topic Exchange — 通配符匹配
 * * 匹配一个单词， # 匹配多个单词
 */
channel.exchangeDeclare("topic_exchange", BuiltinExchangeType.TOPIC);
channel.queueBind("us_orders", "topic_exchange", "order.us.#");
channel.queueBind("cn_orders", "topic_exchange", "order.cn.#");
// "order.us.create" → us_orders
// "order.cn.pay"    → cn_orders

/**
 * 3. Fanout Exchange — 广播
 */
channel.exchangeDeclare("fanout_exchange", BuiltinExchangeType.FANOUT);
channel.queueBind("queue_a", "fanout_exchange", "");  // routing key 被忽略
channel.queueBind("queue_b", "fanout_exchange", "");
// 所有消息发送到所有绑定的队列

/**
 * 4. Headers Exchange — 按消息头匹配
 */
Map<String, Object> headers = new HashMap<>();
headers.put("x-match", "all");  // all = 全部匹配，any = 任意匹配
headers.put("format", "json");
headers.put("type", "order");
channel.queueBind("header_queue", "headers_exchange", "", headers);
// 消息头包含 format=json 且 type=order 才路由到该队列

/**
 * RabbitMQ 工作队列 — 任务分发
 */
// 生产者
public class TaskProducer {
    private static final String QUEUE_NAME = "task_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection conn = factory.newConnection();
             Channel channel = conn.createChannel()) {
            // 持久化队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);

            String message = "任务: " + LocalDateTime.now();
            // 持久化消息
            channel.basicPublish("", QUEUE_NAME,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                message.getBytes(StandardCharsets.UTF_8));
            System.out.println("发送: " + message);
        }
    }
}

// 消费者 — 公平分发 + 手动 ACK
public class Worker {
    private static final String QUEUE_NAME = "task_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection conn = factory.newConnection();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.basicQos(1); // 一次只取一条消息（公平分发）

        DeliverCallback callback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            try {
                System.out.println("处理: " + message);
                Thread.sleep(2000); // 模拟处理耗时
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            } catch (Exception e) {
                channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
            }
        };

        channel.basicConsume(QUEUE_NAME, false, callback, consumerTag -> {});
    }
}

✅ 强项：
  - 复杂路由策略（ Topic/Fanout/Header 多种交换机）
  - 需要灵活的消息路由和过滤
  - 中小流量、低延迟场景
  - 微服务间异步调用
  - 社区活跃、文档丰富、运维工具完善

❌ 弱项：
  - 吞吐量有限（~1万 TPS）
  - 消息堆积后性能急剧下降
  - 不支持事务消息
  - 不支持消息对比
  - Erlang 语言导致二次开发门槛高

/**
 * 1. 高可用架构配置
 */
// broker-a.properties
brokerClusterName=DefaultCluster
brokerName=broker-a
brokerId=0
namesrvAddr=192.168.1.1:9876;192.168.1.2:9876
brokerRole=SYNC_MASTER
flushDiskType=ASYNC_FLUSH
// 多 NameServer 防止单点

/**
 * 2. 消息轨迹
 */
DefaultMQProducer producer = new DefaultMQProducer("trace_group");
producer.setNamesrvAddr("localhost:9876");
producer.setEnableMsgTrace(true); // 开启轨迹

✅ 强项：
  - 高吞吐（~10万 TPS+），适合中大规模业务
  - 原生支持事务消息（RocketMQ 核心优势）
  - 原生支持延时/定时消息
  - 消息轨迹追踪
  - 批量消息、Tag/SQL 过滤
  - 国内生态好，阿里双11 验证
  - Java 原生，二次开发方便

❌ 弱项：
  - 生态不如 Kafka 广泛（流处理方面）
  - 客户端 SDK 质量参差不齐（不同语言版本）
  - 社区活跃度不如 Kafka/RabbitMQ
  - 部署和运维相对复杂
  - 消息路由能力不如 RabbitMQ 灵活

/**
 * 1. 顺序写入磁盘
 */
// Kafka 消息写入是顺序追加到文件末尾
// 顺序写入 ≈ 内存速度（~600MB/s）
// 随机写入 ≈ 磁盘速度（~100 IOPS）

/**
 * 2. 零拷贝（Zero Copy）
 */
// 传统方式（4次数据拷贝）：
// 磁盘 → 内核缓冲区 → 用户缓冲区 → Socket 缓冲区 → 网卡
//
// Kafka 零拷贝（2次数据拷贝）：
// 磁盘 → 内核缓冲区 → 网卡（sendfile 系统调用）
// 减少了两次上下文切换 + 两次数据拷贝

/**
 * 3. 批量压缩与发送
 */
Properties props = new Properties();
props.put("batch.size", 16384);       // 16KB 批次大小
props.put("linger.ms", 5);             // 最多等待 5ms
props.put("compression.type", "snappy"); // 压缩算法
// 批量发送显著提升吞吐量

/**
 * Kafka Streams — 实时计算
 */
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

public class OrderStatisticsStream {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-stats");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> orders = builder.stream("order_topic");

        // 实时计算每分钟的交易额
        orders
            .mapValues(value -> {
                JsonObject order = JsonParser.parseString(value).getAsJsonObject();
                return order.get("amount").getAsDouble();
            })
            .groupByKey()
            .windowedBy(TimeWindows.of(Duration.ofMinutes(1)))
            .aggregate(
                () -> 0.0,
                (key, value, aggregate) -> aggregate + value,
                Materialized.with(Serdes.String(), Serdes.Double())
            )
            .toStream()
            .foreach((key, value) ->
                System.out.println("窗口: " + key + ", 交易额: " + value));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}

✅ 强项：
  - 极致吞吐量（~100万 TPS），大数据首选
  - 持久化 + 高可用，数据不丢
  - 强大的流处理生态（Kafka Streams, ksqlDB）
  - 日志收集、埋点数据、Metrics 指标
  - 日志压缩（Log Compaction），保留最新状态
  - 多语言客户端，社区极其活跃

❌ 弱项：
  - 消息路由能力弱，不支持复杂过滤
  - 事务和可靠性机制不如 RocketMQ 完善
  - 大量 Topic 时性能下降明显
  - 消费者需自己管理偏移量
  - 运维较复杂（依赖 ZooKeeper 或 KRaft）
  - 延迟不是最优（追求吞吐的设计取舍）

/**
 * Redis List 实现简易消息队列
 * 适合小型项目或临时使用
 */
@Component
public class RedisQueueExample {

    @Autowired
    private StringRedisTemplate redis;

    private static final String QUEUE_KEY = "simple:queue";

    /**
     * 生产者：LPUSH 入队
     */
    public void sendMessage(String message) {
        redis.opsForList().leftPush(QUEUE_KEY, message);
    }

    /**
     * 消费者：BRPOP 阻塞消费（支持超时）
     */
    public void consumeMessage() {
        while (true) {
            // BRPOP：阻塞弹出，0 表示永不超时
            List<String> messages = redis.opsForList().rightPop(QUEUE_KEY, 0, TimeUnit.SECONDS);
            process(messages.get(1)); // 返回 [key, value]
        }
    }
}

/**
 * Redis Pub/Sub — 广播消息
 * 注意：消息不持久化，消费者不在线则丢失
 */
@Component
public class RedisPubSubExample {

    @Autowired
    private RedisTemplate<String, String> redis;

    /**
     * 发布消息
     */
    public void publish(String channel, String message) {
        redis.convertAndSend(channel, message);
    }

    /**
     * 订阅消息（需配置 MessageListener）
     */
    @Bean
    public MessageListenerAdapter messageListener() {
        return new MessageListenerAdapter(new RedisMessageSubscriber());
    }

    public static class RedisMessageSubscriber {
        public void handleMessage(String message) {
            System.out.println("收到消息: " + message);
        }
    }
}

✅ 强项：
  - 极简部署，无需额外中间件
  - 微秒级延迟
  - 适合小型项目/工具脚本
  - 团队已有 Redis 基础设施

❌ 弱项：
  - 消息丢失风险（Pub/Sub 不持久化）
  - 无消息确认机制（ACK）
  - 无死信队列、无重试机制
  - 不支持集群消费（广播模式）
  - 堆积性能差（List 过长影响 Redis 性能）
  - 功能极其有限

🟡 建议：
  - 只适合边缘场景（日志采集、非关键通知）
  - 生产环境的业务流程不要使用 Redis 做 MQ