详解RabbitMQ工作模式之发布订阅模式

一、模式核心原理

1.1 核心组件

• Exchange类型：Fanout（消息广播）
• 消息路由规则：完全广播，忽略Routing Key
• 队列绑定机制：动态绑定/解绑

1.2 与传统队列对比

二、环境搭建（Gradle配置）

plugins {
    id 'java'
}

repositories {
    mavenCentral()
}

dependencies {
    implementation 'com.rabbitmq:amqp-client:5.16.0'
    implementation 'org.slf4j:slf4j-api:2.0.7'
    implementation 'ch.qos.logback:logback-classic:1.4.8'
}

三、基础实现代码

3.1 生产者实现

public class Publisher {
    private static final String EXCHANGE_NAME = "news_feed";
    
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            // 声明fanout类型交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            
            String message = "【突发新闻】RabbitMQ 4.0正式发布";
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println(" [x] 发送消息: '" + message + "'");
        }
    }
}

3.2 消费者实现

public class Subscriber {
    private static final String EXCHANGE_NAME = "news_feed";
    
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        
        // 声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        // 创建临时队列
        String queueName = channel.queueDeclare().getQueue();
        // 绑定队列到交换机
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        
        System.out.println(" [*] 等待接收消息...");
        
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x] 收到消息: '" + message + "'");
        };
        
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

四、高级特性实现

4.1 持久化配置

// 持久化交换机
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT, true);

// 持久化队列
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-queue-type", "quorum"); // 使用Quorum队列
channel.queueDeclare("persistent_queue", true, false, false, arguments);

// 持久化消息
AMQP.BasicProperties props = MessageProperties.PERSISTENT_TEXT_PLAIN;
channel.basicPublish(EXCHANGE_NAME, "", props, message.getBytes());

4.2 消息过滤（扩展模式）

// 使用headers交换器实现过滤
Map<String, Object> headers = new HashMap<>();
headers.put("news_type", "sports");

channel.exchangeDeclare("filtered_news", BuiltinExchangeType.HEADERS);
channel.queueBind(queueName, "filtered_news", "", headers);

五、生产环境最佳实践

5.1 集群部署方案

# 组成3节点集群
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app

5.2 监控配置

# Prometheus配置示例
scrape_configs:
  - job_name: 'rabbitmq'
    static_configs:
      - targets: ['rabbitmq:15692']

5.3 性能优化参数

ConnectionFactory factory = new ConnectionFactory();
factory.setRequestedChannelMax(2048); // 最大通道数
factory.setConnectionTimeout(30000);  // 连接超时
factory.setShutdownTimeout(60000);    // 关闭超时
factory.useNio();                     // 使用NIO模式

六、模式应用场景

6.1 实时场景

1. 电商系统：价格变动通知
2. 社交平台：好友动态推送
3. 物联网：设备状态广播

6.2 代码实战：新闻推送系统

// 动态路由实现
public class NewsRouter {
    private static final String EXCHANGE_NAME = "smart_news";
    
    public void routeNews(String category, String content) throws Exception {
        try (Channel channel = connection.createChannel()) {
            String routingKey = "news." + category.toLowerCase();
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, content.getBytes());
        }
    }
}

// 消费者分组
channel.queueDeclare("sports_group", true, false, false, null);
channel.queueBind("sports_group", EXCHANGE_NAME, "news.sports");

七、常见问题解决方案

7.1 消息堆积处理

// 限流配置
channel.basicQos(100); // 每次取100条
channel.basicConsume(queueName, false, deliverCallback, consumerTag -> {});

7.2 消费者宕机恢复

// 消息确认机制
channel.basicConsume(queueName, false, (consumerTag, delivery) -> {
    try {
        processMessage(delivery.getBody());
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
}, consumerTag -> {});

八、性能测试数据

九、扩展应用

9.1 与Spring Boot集成

@Configuration
public class RabbitConfig {
    
    @Bean
    public FanoutExchange newsExchange() {
        return new FanoutExchange("news.fanout");
    }
    
    @Bean
    public Queue autoDeleteQueue1() {
        return new AnonymousQueue();
    }
    
    @Bean
    public Binding binding1(FanoutExchange exchange, Queue autoDeleteQueue1) {
        return BindingBuilder.bind(autoDeleteQueue1).to(exchange);
    }
}

9.2 消息追踪插件

rabbitmq-plugins enable rabbitmq_tracing

十、模式总结

优势：

• 天然支持广播场景
• 消费者动态扩展能力
• 消息路由解耦

适用场景建议：

• 需要1:N消息分发的场景
• 实时性要求高的系统
• 消费者数量动态变化的系统

后续优化方向：

1. 结合MQTT协议实现移动端推送
2. 使用流式传输处理大数据量
3. 集成DLX（死信队列）增强可靠性

完整代码示例已托管至Gitee：https://gitee.com/example/rabbitmq-pubsub-demo

注意事项：

• 生产环境建议开启TLS加密
• 重要消息务必开启持久化
• 建议使用镜像队列保证高可用

欢迎在评论区留下您在实际使用中遇到的问题，我们将挑选典型问题进行详细解答！