您当前的位置： 首页 > 慢生活 > 程序人生 网站首页程序人生

尚硅谷RabbitMQ视频-010-基础篇-工作模式-发布订阅模式

发布时间：2026-10-24 22:11:59编辑：雪饮阅读（）

• Fanout 模型（广播模式）
生产者 ---> Fanout 类型交换机
                                        ---> 队列 1 ---> c1
　　　　　　　　　　　　　　　 ---> 队列 2 ---> c2、c3、c4
　　　　　　　　　　　　　　　 ---> 队列 3 ---> c5
Fanout 是广播：消息发给所有绑定到此交换机的队列
同一个队列绑定多个消费者时，队列内的多个消费者是竞争关系，一条消息只会被其中一个消费者消费
二、交换机 3 种常见类型
Fanout：广播，将消息发送给所有绑定到交换机的队列
Direct：定向，把消息交给符合指定 routing key 的队列
Topic：通配符，把消息交给符合 routing pattern（路由模式）的队列
三、重要注意点Exchange（交换机）只负责转发消息，不具备存储消息的能力
因此如果没有任何队列与 Exchange 绑定，或者没有符合路由规则的队列，那么消息会丢失！


publish/Subscribe 模式说明（RabbitMQ 发布订阅）
组件之间关系
• 生产者把消息发送到交换机
• 队列直接和交换机绑定
工作机制
消息发送到交换机上，就会以广播的形式发送给所有已绑定队列
理解概念
• Publish：发布，这里就是把消息发送到交换机上
• Subscribe：订阅，这里只要把队列和交换机绑定，事实上就形成了一种订阅关系
简单说明：生产者 (P) → 交换机 (X) → 多个绑定的队列（消息广播到全部队列）

X = Exchange（交换机）


Fanout模式的生产者
Producer.java：
package org.example.fanout;

import org.example.workQueues.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.BuiltinExchangeType;

public class Producer {

    public static void main(String[] args) throws Exception {
        // 1、获取连接
        Connection connection = ConnectionUtil.getConnection();

        // 2、创建频道
        Channel channel = connection.createChannel();

        // 参数1. exchange: 交换机名称
        // 参数2. type: 交换机类型
        //      DIRECT("direct"): 定向
        //      FANOUT("fanout"): 扇形（广播），发送消息到每一个与之绑定队列。
        //      TOPIC("topic"): 通配符的方式
        //      HEADERS("headers"): 参数匹配
        // 参数3. durable: 是否持久化(所以说除了队列有持久化，交换机也可以有持久化，RabbitMQ 重启之后交换机还在)
        // 参数4. autoDelete: 自动删除
        // 参数5. internal: 内部使用。一般false
        // 参数6. arguments: 其它参数
        String exchangeName = "test_fanout";

        // 3、创建交换机
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);

        // 4、创建队列
        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";

        channel.queueDeclare(queue1Name,true,false,false,null);
        channel.queueDeclare(queue2Name,true,false,false,null);

        // 5、绑定队列和交换机
        // 参数1. queue: 队列名称
        // 参数2. exchange: 交换机名称
        // 参数3. routingKey: 路由键，绑定规则
        //      如果交换机的类型为fanout，routingKey设置为""
        channel.queueBind(queue1Name,exchangeName,"");
        channel.queueBind(queue2Name,exchangeName,"");

        String body = "日志信息：张三调用了findAll方法...日志级别：info...";
        // 6、发送消息(到交换机)
        channel.basicPublish(exchangeName, "", null,body.getBytes());

        // 7、释放资源
        channel.close();
        connection.close();
    }
}

生产者运行后可以在rabbitmq后台查看发送过去的消息（要rabbitmq所支持的utf8格式）
所以上面生产者这里要修改下发送消息过去的消息体编码
 channel.basicPublish(exchangeName, "", null,body.getBytes(StandardCharsets.UTF_8));


获取消息
然后在rabbitmq的后台中就可以get message

那你可以看到这里取消息默认是Nack message requeue true，也就是并非实际取走。
相当于只是查看下。

这个是 RabbitMQ 管理后台 Get messages 下拉框里的 Ack Mode（确认模式），对应 AMQP 的 basic.nack 操作：
Nack = Negative Acknowledge，否定确认；requeue=true：消息拿出来看完之后，放回原队列RabbitMQ
在页面上点击 Get Message(s) 获取这条消息只是预览查看，看完之后告诉 RabbitMQ：这条消息我没处理好，放回去，别的消费者还能继续消费它。消息不会被删除，会重新回到队列，redelivered 标记会置为 1（代表是重投递消息）

Fanout模式的消费者1
Consumer1.java:
package org.example.fanout;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

import org.example.workQueues.util.ConnectionUtil;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
;

public class Consumer1 {
public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        String queue1Name = "test_fanout_queue1";
        channel.queueDeclare(queue1Name,true,false,false,null);

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //解析生产者生产的utf8编码的字符串
                System.out.println("body:"+new String(body,StandardCharsets.UTF_8));
                System.out.println("队列 1 消费者 1 将日志信息打印到控制台.....");
            }
        };

        channel.basicConsume(queue1Name,true,consumer);
    }
}



Fanout模式的消费者2
Consumer2.java:
package org.example.fanout;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

import org.example.workQueues.util.ConnectionUtil;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

public class Consumer2 {
public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();

        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue2Name,true,false,false,null);

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body:"+new String(body,StandardCharsets.UTF_8));
                System.out.println("队列 2 消费者 2 将日志信息打印到控制台.....");
            }
        };

        channel.basicConsume(queue2Name,true,consumer);
    }
}