RabbitMQ之topic模式

Myth丶恋晨 2022-03-09 03:46 282阅读 0赞

## 概述 ##

消息发送到topic类型的exchange上时不能随意指定routing\_key（一定是指由一系列由点号连接单词的字符串，单词可以是任意的，但一般都会与消息或多或少的有些关联）。Routing key的长度不能超过255个字节。

Binding key也一定要是同样的方式。Topic类型的exchange就像一个直接的交换：一个由生产者指定了确定routing key的消息将会被推送给所有Binding key能与之匹配的消费者。然而这种绑定有两种特殊的情况：

*  \*（星号）：可以（只能）匹配一个单词
 *  \#（井号）：可以匹配多个单词（或者零个）

![222234326373928.png][]

## 例子 ##

生产者

package com.luke.sbdubbo.user.mq.topic;
    
    import com.rabbitmq.client.*;
    
    public class Producer {
    
        public static final String EXCHANGE_NAME = "exchange_topic_demo";
        public static final String QUEUE_NAME1 = "queue_topic_1";
        public static final String QUEUE_NAME2 = "queue_topic_2";
        public static final String  IP_ADDRESS = "192.168.0.114";
        public static final int PORT = 5672;
    
        public static void main(String[] args) throws Exception{
    
            //创建连接工厂
            Address[] addresses = new Address[]{
                    new Address(IP_ADDRESS,PORT)
            };
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUsername("root");
            factory.setPassword("root");
            //获取连接通道
            Connection connection = factory.newConnection(addresses);
            Channel channel = connection.createChannel();
    
            //声明交换器和队列，并生成(实际生成环境，建议事先建好交换器、队列，还有绑定关系)
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC,true);
            channel.queueDeclare(QUEUE_NAME1,true,false,false,null);
            channel.queueDeclare(QUEUE_NAME2,true,false,false,null);
            String bindingKey1 = "*.orange.*";
            String bindingKey2 = "*.*.rabbit";
            String bindingKey3 = "lazy.#";
            //队列绑定交换器
            channel.queueBind(QUEUE_NAME1,EXCHANGE_NAME,bindingKey1);
            channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,bindingKey2);
            channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,bindingKey3);
    
            //路由键
           //String ROUTING_KEY = "aa.orange.bb";
            //String ROUTING_KEY = "aa.orange.rabbit";
            String ROUTING_KEY = "aa.bbb.rabbit";
    
            //发送消息
            for (int i = 1; i <= 10; i++) {
                String message = "rabbitmq topic message"+i;
                channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,
                        MessageProperties.PERSISTENT_TEXT_PLAIN,
                        message.getBytes("UTF-8"));
            }
    
            //关闭连接
            channel.close();
            connection.close();
    
        }
    
    }

消费者1

package com.luke.sbdubbo.user.mq.topic;
    
    import com.rabbitmq.client.*;
    import java.io.IOException;
    import java.util.concurrent.TimeUnit;
    
    public class Queue1Consumer {
    
        public static final String QUEUE_NAME1 = "queue_topic_1";
        public static final String  IP_ADDRESS = "192.168.0.114";
        public static final int PORT = 5672;
    
        public static void main(String[] args) throws Exception{
            //创建连接工厂
            Address[] addresses = new Address[]{
                    new Address(IP_ADDRESS,PORT)
            };
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUsername("root");
            factory.setPassword("root");
            //获取连接通道
            Connection connection = factory.newConnection(addresses);
            Channel channel = connection.createChannel();
            channel.basicQos(64);
    
            //消费
            Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
                    try {
                        System.out.println("recv1 message: "+new String(body));
                        TimeUnit.SECONDS.sleep(1);
                        //客户端消息确认
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
    
            channel.basicConsume(QUEUE_NAME1,false,consumer);
            //关闭连接
           /* TimeUnit.SECONDS.sleep(5);
            channel.close();
            connection.close();*/
    
        }
    
    }

消费者2

package com.luke.sbdubbo.user.mq.topic;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeUnit;
    
    public class Queue2Consumer {
    
        public static final String QUEUE_NAME2 = "queue_topic_2";
        public static final String  IP_ADDRESS = "192.168.0.114";
        public static final int PORT = 5672;
    
        public static void main(String[] args) throws Exception{
            //创建连接工厂
            Address[] addresses = new Address[]{
                    new Address(IP_ADDRESS,PORT)
            };
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUsername("root");
            factory.setPassword("root");
            //获取连接通道
            Connection connection = factory.newConnection(addresses);
            Channel channel = connection.createChannel();
            channel.basicQos(64);
    
            //消费
            Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
                    try {
                        System.out.println("recv2 message: "+new String(body));
                        TimeUnit.SECONDS.sleep(1);
                        //客户端消息确认
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
    
            channel.basicConsume(QUEUE_NAME2,false,consumer);
            //关闭连接
            /*TimeUnit.SECONDS.sleep(5);
            channel.close();
            connection.close();*/
    
        }
    
    }

结果

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3Mzk1NTE_size_16_color_FFFFFF_t_70][]![20190312112521418.png][]

## 参考 ##

[https://www.cnblogs.com/leocook/p/mq\_rabbitmq\_5.html][https_www.cnblogs.com_leocook_p_mq_rabbitmq_5.html]

[https://blog.csdn.net/heyutao007/article/details/50131089][https_blog.csdn.net_heyutao007_article_details_50131089]

[222234326373928.png]: /images/20220309/32730b61c5b745ed84184d1436530479.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3Mzk1NTE_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190312112455683.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA3Mzk1NTE=,size_16,color_FFFFFF,t_70
[20190312112521418.png]: /images/20220309/db166ba71f3d486da5a8983a740369aa.png
[https_www.cnblogs.com_leocook_p_mq_rabbitmq_5.html]: https://www.cnblogs.com/leocook/p/mq_rabbitmq_5.html
[https_blog.csdn.net_heyutao007_article_details_50131089]: https://blog.csdn.net/heyutao007/article/details/50131089