Topic匹配模式

你的名字 2022-06-11 07:37 161阅读 0赞

topic类型的交换器允许在RabbitMQ中使用模糊匹配来绑定自己感兴趣的信息:

\*（星号）表示一个单词  
\#（井号）表示零个或者多个单词

![SouthEast][]

*  RoutingKey为“black.critical.high”的日志会投递到queue1和queue2,。
 *  RoutingKey为“red.critical.high”的日志会只投递到queue2。
 *  RoutingKey为“white.critical.high”的日志会投递到queue2，并且虽然queue2的两个匹配规则都符合但只会向queue2投递一份。

public class Producer {
            private static final String EXCHANGE_NAME = "topic_logs";
            public static void main(String[] argv) {
                Connection connection = null;
                Channel channel = null;
                try {
                    ConnectionFactory factory = new ConnectionFactory();
                    factory.setHost("localhost");
    
                    connection = factory.newConnection();
                    channel = connection.createChannel();
    //			声明一个匹配模式的交换器
                    channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    
                    // 待发送的消息
                    String[] routingKeys = new String[]{"black.critical.high",
                            "red.critical.high",
                            "white.critical.high",
                            };
    //			发送消息
                    for(String severity :routingKeys){
                        String message = "From "+severity+" routingKey' s message!";
                        channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
                        System.out.println("TopicSend [x] Sent '" + severity + "':'" + message + "'");
                    }
    
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    if (connection != null) {
                        try {
                            connection.close();
                        } catch (Exception ignore) {
                        }
                    }
                }
            }
    }

public class Consumer01 {
    
        private static final String EXCHANGE_NAME = "topic_logs";
    
        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();
            // 路由关键字
            String[] routingKeys = new String[]{"#.high", "white.critical.*"};
    //		绑定路由关键字
            for (String bindingKey : routingKeys) {
                channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
                System.out.println("ReceiveLogsTopic2 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + bindingKey);
            }
    
            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("ReceiveLogsTopic2 [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
                }
            };
            channel.basicConsume(queueName, true, consumer);
        }
    }

public class Consumer02 {
        private static final String EXCHANGE_NAME = "topic_logs";
    
        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();
            // 路由关键字
            String[] routingKeys = new String[]{"black.critical.high"};
    //		绑定路由关键字
            for (String bindingKey : routingKeys) {
                channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
                System.out.println("ReceiveLogsTopic1 exchange:"+EXCHANGE_NAME+", queue:"+queueName+", BindRoutingKey:" + bindingKey);
            }
    
            System.out.println("ReceiveLogsTopic1 [*] 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("ReceiveLogsTopic1 [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
                }
            };
            channel.basicConsume(queueName, true, consumer);
        }
    }

参考文章:

http://blog.csdn.net/Super\_RD/article/details/70850453

[SouthEast]: /images/20220611/41e4b992b325432891e66527c6964097.png