RabbitMQ RabbitMQ 学习笔记 -- 04 扇形交换机

jan · 2020-10-24 15:09:07 · 热度: 44

一、介绍

这章我们实现扇形模式将消息传递给多个队列，这种模式也称为“发布/订阅”，

回顾：

生产者是发送消息的用户应用程序。
队列是存储消息的缓冲区。
消费者是接收消息的用户应用程序。

RabbitMQ 消息传递模型中的核心思想是：一直不会将任何消息直接发送到队列上，而且，生产者经常不知道有没有将消息传递到任何队列。

相反，生产者只能将消息发送到交换机。交换消息是一件非常简单的事情。一方面，交换机它接收来自生产者的消息，另一方面，将它们的消息推入队列。交换机必须确切地知道如何处理接收到的消息。是否应将其附加到特定队列？是否应该将其附加到许多队列中？还是应该丢弃它。这些规则都由交换类型定义。

有几种交换类型可用：direct，topic，headers 和 fanout。我们将集中讨论—-扇形。

1.直连型交换机，根据消息携带的路由键将消息投递给对应队列。

大致流程，有一个队列绑定到一个直连交换机上，同时赋予一个路由键 routing key 。
然后当一个消息携带着路由值为X，这个消息通过生产者发送给交换机时，交换机就会根据这个路由值X去寻找绑定值也是X的队列。

在前面部分，我们没有看到使用交换机，但仍然能够将信息发送到队列。因为我们使用的是默认交换机，通过空字符串(“”)进行标识。

private static final String DEFAULT_EXCHANGE = "";

我们之前发布消息使用的是：

rabbitTemplate.convertAndSend(routingKey, msg);

第一个参数是路由密钥，默认情况下，RabbitTemplate 将消息发送到默认交换机。每个队列自动以默认队列名称绑定到默认交换机(AMQP default)。这就是为什么我们可以使用队列的名称作为路由键可以确保消息最终出现在队列中的原因。

2.扇型交换机没有路由键概念，就算你绑了路由键也是无视的。这个交换机在接收到消息后，会直接转发到绑定到它上面的所有队列。

二、web 控制台操作

我们在 web 控制台界面创建一个扇形交换机 rabbit_exchange_fanout

创建两个队列，rabbit.queue.fanout.1 rabbit.queue.fanout.2 并将其绑定在 rabbit_exchange_fanout 上，

①.可以在交换机中进行绑定队列：

②.或者在队列中，绑定到交换机中：

在扇形交换机中，可以随便设置或为空路由键，因为扇型交换机没有路由键概念，就算你绑了路由键也是无视的。这个交换机在接收到消息后，会直接转发到绑定到它上面的所有队列。

绑定成功状态：

我们在交换机中发布消息：

扇形交换机的 routingKey 可填可不填，只要连上就会发送至绑定上的队列

这时候，在队列中我们可以看到消息同时发布在了两台交换机中：

效果：可以看到两条队列消费的消息时一样的：

三、代码方式

声明交换机，队列

@Configuration
public class config {
    // 声明队列名 rabbit.queue.fanout.1
    @Bean
    public Queue queue1() {
        return new Queue("rabbit.queue.fanout.1");
    }
    // 声明队列名 rabbit.queue.fanout.2
    @Bean
    public Queue queue2() {
        return new Queue("rabbit.queue.fanout.2");
    }
    // 声明扇形交换机名 rabbit_exchange_fanout
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("rabbit_exchange_fanout");
    }
    // 将扇形交换机与队列 rabbit.queue.fanout.1 相绑定
    @Bean
    public Binding binding1(FanoutExchange fanoutExchange, Queue queue1) {
        return BindingBuilder.bind(queue1).to(fanoutExchange);
    }
    // 将扇形交换机与队列 rabbit.queue.fanout.2 相绑定
    @Bean
    public Binding binding2(FanoutExchange fanoutExchange, Queue queue2) {
        return BindingBuilder.bind(queue2).to(fanoutExchange);
    }
}

消费者：

@Component
public class consumer {
    @RabbitListener(queues = "rabbit.queue.fanout.1")
    public void receive(String msg) {
        System.out.println("队列 rabbit.queue.fanout.1 收到消息：" + msg);
    }
    @RabbitListener(queues = "rabbit.queue.fanout.2")
    public void receive2(String msg) {
        System.out.println("队列 rabbit.queue.fanout.2 收到消息：" + msg);
    }
}

生产者：
使用 test 来充当生产者发布消息

@Autowired private RabbitTemplate rabbitTemplate;
@Test
public void producer() {
    String exchange = "rabbit_exchange_fanout";
    String routingKey = "";
    String msg = "我不要你觉得，我要我觉得";
    rabbitTemplate.convertAndSend(exchange, routingKey, msg);
}

输出：

队列 rabbit.queue.fanout.1 收到消息：我不要你觉得，我要我觉得
队列 rabbit.queue.fanout.2 收到消息：我不要你觉得，我要我觉得

为您推荐与 rabbitmq 相关的帖子：

0 个赞 0 收藏

暂无回复。

需要登录后方可回复, 如果你还没有账号请点击这里注册。