【RabbitMQ】使用SpringAMQP的Publish/Subscribe(发布/订阅)

Publish/Subscribe

发布(Publish)、订阅(Subscribe)：允许将同一个消息发送给多个消费者

注意：exchange负责消息路由，而不是存储，路由失败则消息丢失

常见的**X(exchange–交换机)***类型：

• Fanout 广播
• Direct 路由
• Topoc 话题

发布订阅–FanoutExchange

案例三：利用SpringAMQP演示广播交换机的使用

1.在消费者（consumer包）中，创建一个FanoutConfig类，声明队列、交换机，并将两者绑定

@ConfigurationpublicclassFanoutConfig{//声明FanoutExchange交换机itcast.fanout@BeanpublicFanoutExchangefanoutExchange(){returnnewFanoutExchange("itcast.fanout");}//声明fanout.queue1队列@BeanpublicQueuefanoutQueue1(){returnnewQueue("fanout.queue1");}//绑定fanout.queue1队列和交换机@BeanpublicBindingbindingQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){returnBindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}//声明fanout.queue2队列@BeanpublicQueuefanoutQueue2(){returnnewQueue("fanout.queue2");}//绑定fanout.queue2队列和交换机@BeanpublicBindingbindingQueue2(Queue fanoutQueue2,FanoutExchange fanoutExchange){returnBindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}}

2.在消费者（consumer包）中，编写两个消费者SpringRabbitListener方法，分别监听fanout.queue1和fanout.queue2

@ComponentpublicclassSpringRabbitListener{@RabbitListener(queues ="fanout.queue1")publicvoidlistenFanoutQueue1(String msg){System.out.println("消费者接收到fanout.queue1的消息：【"+ msg +"】");}@RabbitListener(queues ="fanout.queue2")publicvoidlistenFanoutQueue2(String msg){System.out.println("消费者接收到fanout.queue2的消息：【"+ msg +"】");}}

3.在publiSher中编写测试方法，向itcast.fanout发送消息

@RunWith(SpringRunner.class)@SpringBootTestpublicclassSpringAmqpTest{@AutowiredprivateRabbitTemplate rabbitTemplate;@TestpublicvoidtestSendFanoutExchange(){// 交换机名称String exchangeName ="itcast.fanout";// 消息String message ="hello, every one!";// 发送消息
        rabbitTemplate.convertAndSend(exchangeName,"", message);}}

4.运行，结果

发布订阅–DirectExchange

将接收到的消息根据规则路由到指定的Queue，因此称为路由模式(routes)

• 每一个Queue都与Exchange设置一个BindingKey（暗号）
• 发布者发送消息时，指定消息的RoutingKey
• Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

案例四：利用SpringAMQP演示DirectExchange的使用

1.直接用注解

@RabbitListener声明Exchange、Queue、RoutingKey

@QueueBinding注解用于绑定队列和交换器，并指定路由键

@ComponentpublicclassSpringRabbitListener{@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="direct.queue1"),
            exchange =@Exchange(name ="itcast.direct", type =ExchangeTypes.DIRECT),
            key ={"red","blue"}))publicvoidlistenDirectQueue1(String msg){System.out.println("消费者接收到direct.queue1的消息：【"+ msg +"】");}@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="direct.queue2"),
            exchange =@Exchange(name ="itcast.direct", type =ExchangeTypes.DIRECT),
            key ={"red","yellow"}))publicvoidlistenDirectQueue2(String msg){System.out.println("消费者接收到direct.queue2的消息：【"+ msg +"】");}}

2.测试

@RunWith(SpringRunner.class)@SpringBootTestpublicclassSpringAmqpTest{@AutowiredprivateRabbitTemplate rabbitTemplate;@TestpublicvoidtestSendDirectExchange(){// 交换机名称String exchangeName ="itcast.direct";// 消息String message ="hello, red!";// 发送消息
        rabbitTemplate.convertAndSend(exchangeName,"red", message);}}

3.结果

发布订阅–TopicExchange

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以**

.

**分割

Queue与Exchange指定BindingKey时可以使用通配符

• # : 代指0个或多个单词
• * : 代指一个单词

案例五：利用SpringAMQP演示TopicExchange的使用

1.直接用注解

@RabbitListener声明Exchange、Queue、RoutingKey

@QueueBinding注解用于绑定队列和交换器，并指定路由键

@ComponentpublicclassSpringRabbitListener{@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="topic.queue1"),
            exchange =@Exchange(name ="itcast.topic", type =ExchangeTypes.TOPIC),
            key ="china.#"))publicvoidlistenTopicQueue1(String msg){System.out.println("消费者接收到topic.queue1的消息：【"+ msg +"】");}@RabbitListener(bindings =@QueueBinding(
            value =@Queue(name ="topic.queue2"),
            exchange =@Exchange(name ="itcast.topic", type =ExchangeTypes.TOPIC),
            key ="#.news"))publicvoidlistenTopicQueue2(String msg){System.out.println("消费者接收到topic.queue2的消息：【"+ msg +"】");}}

2.测试

@RunWith(SpringRunner.class)@SpringBootTestpublicclassSpringAmqpTest{@AutowiredprivateRabbitTemplate rabbitTemplate;@TestpublicvoidtestSendTopicExchange(){// 交换机名称String exchangeName ="itcast.topic";// 消息String message ="合理小姐是凑巧先生独一无二的女主角";//String message = "今天天气不错，我的心情好极了!";// 发送消息
         rabbitTemplate.convertAndSend(exchangeName,"china.news", message);//rabbitTemplate.convertAndSend(exchangeName, "china.weather", message);}}

3.结果