IT评测·应用市场-qidao123.com

标题: RabbitMQ---交换机-Fanout-Direct [打印本页]

---

作者: 金歌    时间: 2024-7-22 23:55
标题: RabbitMQ---交换机-Fanout-Direct

• Publisher：生产者，不再发送消息到队列中，而是发给交换机
  
• Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处置惩罚消息，例如递交给某个特别队列、递交给所有队列、或是将消息抛弃。到底如何操作，取决于Exchange的范例。
  
• Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列肯定要与交换机绑定。
  
• Consumer：斲丧者，与以前一样，订阅队列，没有变化
  

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！
交换机的范例有四种：

• Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
  
• Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列
  
• Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符
  
• Headers：头匹配，基于MQ的消息头匹配，用的较少。
  

Fanout交换机

简朴点来说，就是生产者把消息发给交换机，交换机根据路由（绑定规则）来转发消息给队列，斲丧者订阅队列，获得消息。
Fanout，英文翻译是扇出，我觉得在MQ中叫广播更合适。
在广播模式下，消息发送流程是这样的：

• 1） 可以有多个队列
  
• 2） 每个队列都要绑定到Exchange（交换机）
  
• 3） 生产者发送的消息，只能发送到交换机
  
• 4） 交换机把消息发送给绑定过的所有队列
  
• 5） 订阅队列的斲丧者都能拿到消息
  注意：我下面的代码都是在上一个加依靠的底子上的，可看我上一个文档
  

01使用官方文档的

消息发送：

2. import com.rabbitmq.client.Channel;
3. import com.rabbitmq.client.Connection;
4. import com.rabbitmq.client.ConnectionFactory;
6. import java.util.Scanner;
8. public class EmitLog {
9.     //定义交换机
10.     private static final String EXCHANGE_NAME = "fanout-exchange";
11.     public static void main(String[] argv) throws Exception {
12.         //创建连接工厂
13.         ConnectionFactory factory = new ConnectionFactory();
14.         factory.setHost("localhost");
15.         //建立连接和通道
16.         try (Connection connection = factory.newConnection();
17.              Channel channel = connection.createChannel()) {
18.             //通过channel.exchangeDeclare方法声明了一个名为EXCHANGE_NAME（即"logs"）的交换机
19.             // 并指定了其类型为fanout。fanout类型的交换机会将消息广播到所有与之绑定的队列中。
20.             channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
21.            //发送消息
22.             Scanner scanner=new Scanner(System.in);
23.             while(scanner.hasNext()){
24.                 String message = scanner.nextLine();
25.                 //channel.basicPublish方法将消息发布到前面声明的交换机中。
26.                 // 注意，这里的routingKey（即第二个参数）为空字符串""，因为对于fanout类型的交换机来说，routingKey是不起作用的。
27.                 channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
28.                 System.out.println(" [x] Sent '" + message + "'");
29.             }
30.             }
32.     }
33. }

复制代码

接收：

注意肯定要创建队列，不然只有交换机没用

1. import com.rabbitmq.client.Channel;
2. import com.rabbitmq.client.Connection;
3. import com.rabbitmq.client.ConnectionFactory;
4. import com.rabbitmq.client.DeliverCallback;
6. public class ReceiveLogs {
7.     private static final String EXCHANGE_NAME = "fanout-exchange";
9.     public static void main(String[] argv) throws Exception {
10.         ConnectionFactory factory = new ConnectionFactory();
11.         factory.setHost("localhost");
12.         Connection connection = factory.newConnection();
13.         //创建通道
14.         Channel channel1 = connection.createChannel();
15.         Channel channel2 = connection.createChannel();
16.         //声明交换机
17.         channel1.exchangeDeclare(EXCHANGE_NAME, "fanout");
18.         channel2.exchangeDeclare(EXCHANGE_NAME, "fanout");
19.         //队列的名字
20.         String queueName = "星星";
21.         //创建队列
22.         channel1.queueDeclare(queueName, true, false, false, null);
23.         channel1.queueBind(queueName, EXCHANGE_NAME, "");
24.         String queueName1 = "晨晨";
25.         //创建队列
26.         channel2.queueDeclare(queueName1, true, false, false, null);
27.         channel2.queueBind(queueName1, EXCHANGE_NAME, "");
28.         System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
30.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
31.             String message = new String(delivery.getBody(), "UTF-8");
32.             System.out.println(" [1] Received '" + message + "'");
33.         };
34.         DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {
35.             String message = new String(delivery.getBody(), "UTF-8");
36.             System.out.println(" [2] Received '" + message + "'");
37.         };
38.         channel1.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
39.         channel2.basicConsume(queueName1, true, deliverCallback2, consumerTag -> { });
40.     }
41. }

复制代码

可以看到fanout范例，生产者发送一个消息，所有的斲丧者都能接收到，这个范例不消设置路由
02注解形式的：

消息发送

先在配置类中声明交换机，队列，以及绑定关系

1.    public static final String FANOUT_QUEUE_1 = "fanout.queue.1";
2.     public static final String FANOUT_QUEUE_2 = "fanout.queue.2";
3.     public static final String FANOUT_EXCHANGE = "fanout.exchange";
6.     @Bean
7.     public Queue fanoutQueue1() {
8.         return new Queue(FANOUT_QUEUE_1);
9.     }
11.     @Bean
12.     public Queue fanoutQueue2() {
13.         return new Queue(FANOUT_QUEUE_2);
14.     }
16.     @Bean
17.     public FanoutExchange fanoutExchange() {
18.         return new FanoutExchange(FANOUT_EXCHANGE);
19.     }
21.     @Bean
22.     public Binding binding1() {
23.         return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
24.     }
26.     @Bean
27.     public Binding binding2() {
28.         return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
29.     }

复制代码

Test类中添加测试方法：

1. @Test
2. public void testFanoutExchange() {
3.     // 交换机名称
4.     String exchangeName = "fanout.exchange";
5.     // 消息
6.     String message = "hello, everyone!";
7.     rabbitTemplate.convertAndSend(exchangeName, "", message);
8. }

复制代码

消息接收

在添加两个方法，作为斲丧者：

1. @RabbitListener(queues = "fanout.queue.1")
2. public void listenFanoutQueue1(String msg) {
3.     System.out.println("消费者1接收到Fanout消息：【" + msg + "】");
4. }
6. @RabbitListener(queues = "fanout.queue.2")
7. public void listenFanoutQueue2(String msg) {
8.     System.out.println("消费者2接收到Fanout消息：【" + msg + "】");
9. }

复制代码

总结

交换机的作用是什么？

• 接收publisher发送的消息
  
• 将消息按照规则路由到与之绑定的队列
  
• 不能缓存消息，路由失败，消息丢失
  
• FanoutExchange的会将消息路由到每个绑定的队列
  

Direct交换机

在Fanout模式中，一条消息，会被所有订阅的队列都斲丧。但是，在某些场景下，我们希望不同的消息被不同的队列斲丧。这时就要用到Direct范例的Exchange。

在Direct模型下：

• 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
  
• 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。
  
• Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判定，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息
  

此处我省略了官方文档那种，直接springboot注解那种的，而且不再bean。而是全注解那种的，也是对黑马代码的进一步优化，不手动操作添加
下面是黑马的案例
案例需求如图：

• 声明一个名为hmall.direct的交换机
  
• 声明队列direct.queue1，绑定hmall.direct，bindingKey为blud和red
  
• 声明队列direct.queue2，绑定hmall.direct，bindingKey为yellow和red
  
• 在consumer服务中，编写两个斲丧者方法，分别监听direct.queue1和direct.queue2
  
• 在publisher中编写测试方法，向hmall.direct发送消息
  

先在配置类加入下面的然后启动一下：

1.     /** 基于注解的来声明交换机和队列及其绑定关系 */
2.     @RabbitListener( bindings = @QueueBinding(
3.             exchange = @Exchange(name = "heima.direct", type = ExchangeTypes.DIRECT),
4.             value = @org.springframework.amqp.rabbit.annotation.Queue(name = "direct.queue1"),
5.             key = {"red", "blue"}
6.     ))
7.     public void rabbitListener5(String message) {
8.         System.out.println("红蓝: " + message);
9.     }
11.     @RabbitListener( bindings = @QueueBinding(
12.             exchange = @Exchange(name = "heima.direct", type = ExchangeTypes.DIRECT),
13.             value = @org.springframework.amqp.rabbit.annotation.Queue(name = "direct.queue2"),
14.             key = {"yellow","red"}
15.     ))
16.     public void rabbitListener6(String message) {
17.         System.out.println("黄红: " + message);
18.     }

复制代码

消息发送

在Test类中添加测试方法：

1.     @Test
2.     public void testSendDirectExchange() {
3.         // 交换机名称
4.         String exchangeName = "heima.direct";
5.         // 消息
6.         String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
7.         // 发送消息
8.         rabbitTemplate.convertAndSend(exchangeName, "red", message);
9.     }
10.     @Test
11.     public void testSendDirectExchange01() {
12.         // 交换机名称
13.         String exchangeName = "heima.direct";
14.         // 消息
15.         String message = "最新报道，哥斯拉是居民自治巨型气球，虚惊一场！";
16.         // 发送消息
17.         rabbitTemplate.convertAndSend(exchangeName, "blue", message);
18.     }

复制代码

消息接收：

1.     @RabbitListener(queues = "direct.queue1")
2.     public void listenDirectQueue1(String msg) {
3.         System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
4.     }
6.     @RabbitListener(queues = "direct.queue2")
7.     public void listenDirectQueue2(String msg) {
8.         System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
9.     }

复制代码

先点击测试的谁人red的运行一下，在启动项目：

我们再切换为blue这个key：
你会发现，只有斲丧者1收到了消息：

###总结
描述下Direct交换机与Fanout交换机的差别？

• Fanout交换机将消息路由给每一个与之绑定的队列
  
• Direct交换机根据RoutingKey判定路由给哪个队列
  
• 如果多个队列具有相同的RoutingKey，则与Fanout功能类似
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/)

Powered by Discuz! X3.4