Java操作RabbitMQ：交换机类型，Direct、Topic、Fanout​

1. 前言：消息中间件与交换机的角色

在日常开发中，系统间的通信就像城市里的快递网络。RabbitMQ作为AMQP协议的代表性消息中间件，其**交换机（Exchange）**相当于快递分拣中心，决定了消息的传递方向。本文将手把手带您掌握三种核心交换机模式（Direct、Topic、Fanout）在Java中的具体实现和工程应用。

2. 环境准备与技术栈说明

本文示例基于以下技术栈：

• Java 11
• Spring Boot 2.7.x
• RabbitMQ 3.11.x（需提前在本地或服务器安装）
• Lombok（简化POJO类编写）
• JUnit 5（测试用例）

POM依赖片段：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>

3. Direct Exchange（直连交换机）

3.1 运行机制与特征

如同快递柜的固定格口，Direct交换机通过精确匹配路由键投递消息。当绑定队列的路由键与消息路由键完全相同时才会被接收。

3.2 生产与消费完整示例

// 配置类：DirectExchangeConfig
@Configuration
public class DirectExchangeConfig {
    // 声明直连交换机
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange("ORDER_DIRECT_EXCHANGE");
    }

    // 订单支付队列
    @Bean
    public Queue paymentQueue() {
        return new Queue("ORDER.PAYMENT");
    }

    // 绑定队列到交换机
    @Bean
    public Binding paymentBinding() {
        return BindingBuilder.bind(paymentQueue())
               .to(orderExchange()).with("payment.status");
    }
}

// 生产者：OrderService
@Service
public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendPaymentSuccessEvent(String orderId) {
        String message = "订单：" + orderId + " 支付成功";
        // 发送到直连交换机，路由键必须与绑定键一致
        rabbitTemplate.convertAndSend("ORDER_DIRECT_EXCHANGE", 
                    "payment.status", 
                    message);
    }
}

// 消费者：PaymentListener
@Component
@RabbitListener(queues = "ORDER.PAYMENT")
public class PaymentListener {
    
    @RabbitHandler
    public void processPaymentEvent(String message) {
        System.out.println("【支付成功处理】接收到消息：" + message);
        // 处理支付后续逻辑：更新订单状态、发送通知等
    }
}

3.3 典型应用场景

• 订单状态变更通知（支付、发货、完成等）
• 精准推送系统告警信息
• 会员等级变更实时同步

4. Topic Exchange（主题交换机）

4.1 路由规则解析

Topic交换机采用类似文件路径的模式匹配规则：

• #：匹配0个或多个单词（类似通配符）
• *：精确匹配1个单词
• 单词间用.分隔

4.2 完整代码演示

// 配置类：TopicExchangeConfig
@Configuration
public class TopicExchangeConfig {
    // 定义日志主题交换机
    @Bean
    public TopicExchange logExchange() {
        return new TopicExchange("LOG_TOPIC_EXCHANGE");
    }

    // 各服务日志队列
    @Bean Queue userLogQueue() { return new Queue("LOG.USER"); }
    @Bean Queue productLogQueue() { return new Queue("LOG.PRODUCT"); }

    // 带通配符的绑定
    @Bean
    public Binding userBinding() {
        // 匹配 user相关的所有操作日志
        return BindingBuilder.bind(userLogQueue())
               .to(logExchange()).with("log.user.*");
    }

    @Bean
    public Binding productBinding() {
        // 捕获product服务的关键操作
        return BindingBuilder.bind(productLogQueue())
               .to(logExchange()).with("log.product.#");
    }
}

// 日志生产者示例
public class LogProducer {
    public void sendUserLoginLog(Long userId) {
        String routingKey = "log.user.login";
        String message = "用户登录：" + userId;
        rabbitTemplate.convertAndSend("LOG_TOPIC_EXCHANGE", 
                    routingKey, message);
    }

    public void sendProductUpdateLog(Long productId) {
        String routingKey = "log.product.operation.update";
        rabbitTemplate.convertAndSend("LOG_TOPIC_EXCHANGE", 
                    routingKey, "商品更新：" + productId);
    }
}

// 用户日志消费者
@Component
@RabbitListener(queues = "LOG.USER")
public class UserLogConsumer {
    
    @RabbitHandler
    public void handleUserLog(String log) {
        System.out.println("用户服务日志记录 >> " + log);
    }
}

4.4 适用业务场景

• 分布式系统日志分类收集
• 多维度事件通知（如：华东地区.门店.库存变更）
• 复杂业务状态广播（订单.退货.审核通过）

5. Fanout Exchange（广播交换机）

5.1 无差别广播机制

Fanout交换机像电台广播，将消息无条件复制到所有绑定队列，忽略路由键设置。

5.2 代码实操示范

// 配置类：FanoutConfig
@Configuration
public class FanoutConfig {
    // 声明广播交换机
    @Bean
    public FanoutExchange noticeExchange() {
        return new FanoutExchange("NOTICE_FANOUT_EXCHANGE");
    }

    // 三个子系统通知队列
    @Bean public Queue smsQueue() { return new Queue("NOTICE.SMS"); }
    @Bean public Queue emailQueue() { return new Queue("NOTICE.EMAIL"); }
    @Bean public Queue appPushQueue() { return new Queue("NOTICE.APP"); }

    // 所有队列都绑定到同一个广播交换机
    @Bean
    public Binding smsBinding() {
        return BindingBuilder.bind(smsQueue()).to(noticeExchange());
    }
    // 其他绑定类似省略...
}

// 公告生产者示例
public class NoticeService {
    public void publishSystemMaintenanceNotice() {
        String message = "系统将于今晚00:00进行维护升级";
        // 无需指定路由键
        rabbitTemplate.convertAndSend("NOTICE_FANOUT_EXCHANGE", 
                    "", message);
    }
}

// 短信通知消费者
@Component
@RabbitListener(queues = "NOTICE.SMS")
public class SmsNoticeListener {
    
    @RabbitHandler
    public void handleSmsNotice(String notice) {
        System.out.println("短信平台处理：" + notice);
    }
}

5.3 常见使用场景

• 全局系统通知（维护公告、配置更新）
• 多系统数据同步（用户注册后同步到CRM、ERP）
• 实时数据镜像（库存变动广播）

6. 技术选型对比与性能分析

性能优化建议：

• Direct模式适合高吞吐场景（如秒杀订单）
• Topic路由键层级建议不超过4级
• Fanout绑定队列数超过10个时建议分片

7. 开发注意事项与避坑指南

1. 路由键设计规范

   • 采用业务域.操作类型.区域的结构化命名（如：order.payment.success）
   • 避免使用特殊字符（如*、#保留字符）

2. 消息确认机制

// 开启发送确认
spring.rabbitmq.publisher-confirm-type=correlated
// 开启返回模式（路由失败处理）
spring.rabbitmq.publisher-returns=true

// 回调配置示例
@PostConstruct
private void setupConfirm() {
    rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
        if (!ack) {
            log.error("消息未到达交换机：" + cause);
        }
    });
}

3. 消费者幂等性设计

// 使用Redis实现幂等校验
@RabbitHandler
public void processMessage(@Payload String msg, 
                         @Header(AmqpHeaders.MESSAGE_ID) String messageId) {
    if (redisTemplate.hasKey("msg:" + messageId)) {
        return; // 已处理过
    }
    // 业务处理...
    redisTemplate.opsForValue().set("msg:" + messageId, "processed", 30, TimeUnit.MINUTES);
}

8. 总结与场景决策树

当面临交换机选型时，可参考以下决策路径：

是否所有消费者都需要消息？
├── 是 → Fanout模式
└── 否 → 是否需要精确匹配？
           ├── 是 → Direct模式
           └── 否 → Topic模式（支持通配符）