RabbitMQ的Exchange 模式之topic（主题模式）

女爷i 2022-06-04 01:11 160阅读 0赞

任何发送到Topic Exchange的消息都会被转发到所有关联RouteKey中指定“topic”的队列上上

1.  这种模式下需要RouteKey，客户端要提前绑定Exchange与Queue。
2.  如果Exchange没有发现能够与RouteKey匹配的Queue，则会抛弃此消息。
3.  客户端在进行绑定时，要提供一个该队列“感兴趣”的主题，如“\#.log.\#”表示该队列关心所有涉及log的消息(一个RouteKey为”MQ.log.error”的消息会被转发到该队列)。
4.  “\#”表示0个或若干个关键字，“*”表示一个关键字。如“log.*”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.\#”能与上述两者匹配。

![这里写图片描述][python-five.png]  
简单解释一下上图，一个消息产生者P，一个模式为topic的Exchange（交换机），两个队列，两个消息消费者。  
  
在topic模式下：  
  
如果一个消息的 routing key设置 `"quick.orange.rabbit"`  
  
即如下

channel.basicPublish(EXCHANGE_NAME, "quick.orange.rabbit", null, "hello".getBytes());

那么该消息将被转发（推送）至Q1、Q2两个队列  
  
如果一个消息的 routing key设置 `"quick.orange.fox"`  
  
那么该消息将只会被转发（推送）至Q1队列  
  
如果一个消息的 routing key设置 `"quick.orange.fox"`  
  
那么该消息将只会被转发（推送）至Q2队列  
  
当然，如果一个消息的 routing key设置 `" quick.brown.fox "`  
  
那么该消息将不会被转发（推送）到任何队列，它将会丢失，即丢弃该消息  
  
前面提到客户端要提前绑定Exchange与Queue

channel.exchangeDeclare(EXCHANGE_NAME, "topic");

并且需要获取一个新的空队列，用来与EXCHANGE进行binding

String queueName = channel.queueDeclare().getQueue();

然后设置你关心的主题

channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*");

如此这般，这个客户端就只会获取到自己“感兴趣”的主题对应的消息  
  
栗子：  
服务端推送主题如下的消息，即routingKey分别为如下值

"adtec.laoliu","ali.xiaowang", "tencent.laozhang", "adtec.xiaohu"

package com.adtec.rabbitmq;
    
    
    import com.rabbitmq.client.*;
    
    public class EmitLogTopic { 
    
        private static final String EXCHANGE_NAME = "topic";
        private static final String[] routingKeys = { "adtec.laoliu","ali.xiaowang", "tencent.laozhang", "adtec.xiaohu" };
    
        public static void main(String[] argv) throws Exception {
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = factory.newConnection();
    
            Channel channel = connection.createChannel();
    
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    
            String message = getMessage("hello");
    
            //分别推送"adtec.laoliu","ali.xiaowang", "tencent.laozhang", "adtec.xiaohu"消息
            for (String routingKey : routingKeys) {
                String msg = message+ "' from "+routingKey;
                channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());
                System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
            }
            channel.close();
            connection.close();
    
        }
    
        private static String getMessage(String strings) {
    
            return "info: Hello World!";
        }
    }

客户端只对adtec相关的主题感兴趣，即routingKey值为`#.adtec.#`

package rabbitmq;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class ReceiveLogsTopic { 
      private static final String EXCHANGE_NAME = "topic";
      private static final String[] bindingKeys = { "adtec.laoliu", "adtec.xiaowang","adtec.xiaozhou","adtec.xiaohu" };
      public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
    
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = channel.queueDeclare().getQueue();
    
        if (bindingKeys.length < 1) {
          System.err.println("Usage: ReceiveLogsTopic [binding_key]...");
          System.exit(1);
        }
    
        //binding那些bindingKey
        //可以单个单个设置，也可以批量设置
        for (String bindingKey : bindingKeys) {
          channel.queueBind(queueName, EXCHANGE_NAME, "#.adtec.#");
        }
    
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
        Consumer consumer = new DefaultConsumer(channel) {
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope,
                                     AMQP.BasicProperties properties, byte[] body) throws IOException {
            String message = new String(body, "UTF-8");
            System.out.println(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
          }
        };
        channel.basicConsume(queueName, true, consumer);
      }
    }

测试，先运行客户端，再运行服务端  
  
服务端发布的消息  
  
![这里写图片描述][SouthEast]  
  
客户端订阅的消息  
  
![这里写图片描述][SouthEast 1]

[python-five.png]: http://www.rabbitmq.com/img/tutorials/python-five.png
[SouthEast]: /images/20220604/60956e5a53904ee788152bcc993027d8.png
[SouthEast 1]: /images/20220604/be5ac0124d3d4667a42908c32a7cc8c8.png