RabbitMQ（四）：通配符模式

古城微笑少年丶 2021-09-24 15:44 360阅读 0赞

### RabbitMQ：通配符模式 ###

*   *  通配符模式(Topic)
     *  二、使用步骤
     *   *  引入库
         *  生产者
         *  消费者1
         *  消费者2

## 通配符模式(Topic) ##

> 路由模式，routingKey采用通配符模式。

![通配符模式][20200906101745350.png_pic_center]

特点：

*  跟routing模式相比，Topics模式是**通配符**匹配

* : 只能匹配一个词，例如: info.*，他可以匹配 info.hello，info.world
    # : 可以匹配多个词，例如: info.#，它可以匹配 info.hello.world，info.xx，info
    # 可以匹配空

## 二、使用步骤 ##

### 引入库 ###

<dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
    </dependency>

### 生产者 ###

> 两个队列 支付宝 和 微信， 有些人想要支付宝，有些人想要微信 ，而有些人全都要

public class TopicProduct { 
        // 两个队列 支付宝 和 微信， 有些人想要支付宝，有些人想要微信 有些人全都要
        private final static String QUEUE_1 = "queue_topic_alipay";
        private final static String QUEUE_2 = "queue_topic_wechat";
        private final static String EXCHANGE_NAME = "exchange_topic";
    
        /** * 可以匹配 info.alipay,info.alipay.wechat... * 当 info.alipay.wechat 可以同时匹配两个路由key */
        private final static String ROUTINGKEY_1 = "info.#.alipay.#";
        private final static String ROUTINGKEY_2 = "info.#.wechat.#";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            // 连接
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("106.**.**.82");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = null;
            Channel channel = null;
            try{ 
                connection = factory.newConnection();
                channel = connection.createChannel();
                // 声明交换机
                channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
                // 声明队列
                channel.queueDeclare(QUEUE_1,false,false,false,null);
                channel.queueDeclare(QUEUE_2,false,false,false,null);
    
                /* 交换机队列绑定 * 1 queue 队列名称 * 2 exchange 交换机名称 * 3 routingKey 路由Key */
                channel.queueBind(QUEUE_1,EXCHANGE_NAME,ROUTINGKEY_1);
                channel.queueBind(QUEUE_2,EXCHANGE_NAME,ROUTINGKEY_2);
    
                String message = "";
    
                // 参数：String exchange, String routingKey, BasicProperties props, byte[] body
                /** * exchange:交换机 如果不指定("")，就默认交换机 * routingKey：路由key；交换机根据路由key将消息转发到指定的队列，如果使用默认交换机，routingKey为队列名称 * props:额外属性 * body：消息内容 */
                message = "这个是给喜欢 AliPay 的朋友们的~";
                channel.basicPublish(EXCHANGE_NAME,"info.alipay",null,message.getBytes());
                System.out.println(new Date()+":发送了message："+message+", Topic routingKey: "+ROUTINGKEY_1);
    
                message = "喜欢 wechat 的朋友，接好了";
                channel.basicPublish(EXCHANGE_NAME,"info.wechat",null,message.getBytes());
                System.out.println(new Date()+":发送了："+message+", Topic routingKey:"+ROUTINGKEY_2);
    
                message = "同时喜欢 ALIPAY 和 WECHAT 的朋友接好了";
                channel.basicPublish(EXCHANGE_NAME,"info.alipay.wechat",null,message.getBytes());
                System.out.println(new Date()+":发送了："+message+",Topic routingKeys:"+ROUTINGKEY_2+","+ROUTINGKEY_1);
    
            } catch (TimeoutException | IOException e) { 
                e.printStackTrace();
            } finally { 
                channel.close();
                connection.close();
            }
        }
    }

### 消费者1 ###

> 喜欢alipay的

public class Consumer_Topic_1 { 
        private final static String QUEUE_1 = "queue_topic_alipay";
        //private final static String QUEUE_2 = "queue_topic_wechat";
        private final static String EXCHANGE_NAME = "exchange_topic";
        private final static String ROUTINGKEY_1 = "info.#.alipay.#";
        //private final static String ROUTINGKEY_2 = "info.#.wechat.#";
    
        public static void main(String[] args) { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("106.**.**.82");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("root");
            factory.setPassword("root");
    
            Connection connection = null;
            Channel channel = null;
            try { 
                connection = factory.newConnection();
                channel = connection.createChannel();
                // 声明交换机
                channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
                // 声明队列
                channel.queueDeclare(QUEUE_1,false,false,false, null);
    
                // 交换机队列绑定
                channel.queueBind(QUEUE_1,EXCHANGE_NAME,ROUTINGKEY_1);
    
                DefaultConsumer consumer = new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { 
                        String message = new String (body, StandardCharsets.UTF_8);
                        System.out.println("receive:" + message);
                    }
                };
    
                // 接收消息 监听队列
                // 参数：String queue, boolean autoAck, Consumer callback
                /** * queue:队列 * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复 * callback:消费方法，当消费者接收消息执行的方法。 */
                channel.basicConsume(QUEUE_1,true,consumer);
    
            }catch (Exception e){ 
    			e.printStackTrace();
            }
        }
    }

输出如下：

receive:这个是给喜欢 AliPay 的朋友们的~
    receive:喜欢 ALIPAY 和 WECHAT 接好了

### 消费者2 ###

> 喜欢wechat

public class Consumer_Topic_2 { 
        // private final static String QUEUE_1 = "queue_topic_alipay";
        private final static String QUEUE_2 = "queue_topic_wechat";
        private final static String EXCHANGE_NAME = "exchange_topic";
    
        // private final static String ROUTINGKEY_1 = "info.#.alipay.#";
        private final static String ROUTINGKEY_2 = "info.#.wechat.#";
    
        public static void main(String[] args) { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("106.**.**.82");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("root");
            factory.setPassword("root");
    
            Connection connection = null;
            Channel channel = null;
            try { 
                connection = factory.newConnection();
                channel = connection.createChannel();
                // 声明交换机
                channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
                // 声明队列
                channel.queueDeclare(QUEUE_2, false, false, false,null);
    
                // 交换机队列绑定
                /** * 1 queue 队列名称 * 2 exchange 交换机名称 * 3 routingKey 路由Key */
                channel.queueBind(QUEUE_2,EXCHANGE_NAME,ROUTINGKEY_2);
    
                DefaultConsumer consumer = new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) { 
                        String message = new String (body, StandardCharsets.UTF_8);
                        System.out.println("receive:" + message);
                    }
                };
    
                // 接收消息 监听队列
                // 参数：String queue, boolean autoAck, Consumer callback
                /** * queue:队列 * autoAck:自动回复：当消费者接收到消息后，告诉mq消息已经接收。TRUE：自动回复，false：编程回复 * callback:消费方法，当消费者接收消息执行的方法。 */
                channel.basicConsume(QUEUE_2,true,consumer);
    
            }catch (Exception ignored){ 
                ignored.printStackTrace();
            }
        }
    }

输出如下：

receive:喜欢 wechat 的朋友，接好了
    receive:喜欢 ALIPAY 和 WECHAT 接好了

[20200906101745350.png_pic_center]: /images/20210923/c9b9d1f2fab44546a0fba4f54d5822a3.png