RabbitMQ - 第一天:快速入门

我就是我 2024-03-23 13:14 73阅读 0赞

#### RabbitMQ快速入门 ####

*  前言
 *  一、RabbitMQ快速入门
 *   *  1.MQ定义
     *   *  1.大致流程:
     *  2.消息队列模型
     *   *  2.1 模型讲解
 *  二、为什么使用RabbitMQ
 *   *  1.使用RabbitMQ的好处
     *   *  1.1可将一些不需要立即返回接口的业务放到MQ中进行异步处理,这样做可以增加服务器阈值的量,不用怕服务器被瞬间的并发量而冲垮.
     *  使用场景
     *  2.AMQP 和 JMS
     *   *  AMQP
         *  JMS
         *  AMQP和JMS的区别
         *  目前市面上流行的MQ
 *  三.快速入门-简单入门案例
 *   *  1.添加依赖
     *  2.编写生成者
     *  3.编写消费者
     *  4.生产者流转过程
     *   *   *  1.客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。
             *  2.客户端调用connection.createChannel方法。此方法开启信道，其包装的channel.open命令发送给Broker,等待channel.basicPublish方法，对应的AMQP命令为Basic.Publish,这个命令包含了content Header 和content Body()。content Header 包含了消息体的属性，例如:投递模式，优先级等，content Body 包含了消息体本身。
             *  3.客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。
     *  5.消费者流转过程
     *   *   *  1. 消费者客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。
             *  2.消费者客户端调用connection.createChannel方法。和生产者客户端一样，协议涉及Channel . Open/Open-Ok命令。
             *  3. 在真正消费之前，消费者客户端需要向Broker 发送Basic.Consume 命令(即调用channel.basicConsume 方法〉将Channel 置为接收模式，之后Broker 回执Basic . Consume - Ok 以告诉消费者客户端准备好消费消息。
             *  4. Broker 向消费者客户端推送(Push) 消息，即Basic.Deliver 命令，这个命令和Basic.Publish 命令一样会携带Content Header 和Content Body。
             *  5.消费者接收到消息并正确消费之后，向Broker 发送确认，即Basic.Ack 命令。
             *  6. 客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。
 *  四.RabbitMQ工作模式
 *   *  1.Work queues工作队列模式
     *   *  1.1代码示例
     *  2.Publish/Subscribe 发布订阅模式
     *   *   *   *  模式简介
         *  代码示例
         *   *  生产者
             *  消费者
             *  测试结构
     *  3.Routing路由模式
     *   *  模式介绍
         *   *  发送方向交换机发送消息时，必须指定消息的RoutingKey
             *  交换机不再把消息交给每一个绑定的队列，而是根据消息的RoutingKey判断筛选，只有队列的RoutingKey与消息的RoutingKey一致，才会接收到消息
         *  ■生产者
         *  ■消费者
     *  4.Topics通配符模式
     *   *  模式介绍
         *  代码示例
         *   *  ■生产者
             *  ■消费者

--------------------

## 前言 ##

RabbitMQ快速入门系列。

--------------------

## 一、RabbitMQ快速入门 ##

### 1.MQ定义 ###

#### 1.大致流程: ####

消费者(consumer)监听队列内容,当生产者(product)向队列中发布新消息,消费者监听到有新消息需要进行消费的消息,执行对应的逻辑.

### 2.消息队列模型 ###

![在这里插入图片描述][bdc6632f6af346f09f52c2c251698181.png]

#### 2.1 模型讲解 ####

.Broker：标识消息队列服务器实体

.Virtual Host：虚拟主机。标识一批交换机、消息队列和相关对象

.Exchange：交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。

.Queue：消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

.Banding：绑定，用于消息队列和交换机之间的关联。

.Channel：信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟链接，AMQP命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说，建立和销毁TCP都是非常昂贵的开销，所以引入了信道的概念，以复用一条TCP连接。

.Connection：网络连接，比如一个TCP连接。

.Publisher：消息的生产者，也是一个向交换器发布消息的客户端应用程序。

.Consumer：消息的消费者，表示一个从一个消息队列中取得消息的客户端应用程序。

## 二、为什么使用RabbitMQ ##

### 1.使用RabbitMQ的好处 ###

#### 1.1可将一些不需要立即返回接口的业务放到MQ中进行异步处理,这样做可以增加服务器阈值的量,不用怕服务器被瞬间的并发量而冲垮. ####

### 使用场景 ###

**任务异步处理**:将需要大量处理时间的任务放到队列中去执行,减少用户等待时间,只需要执行完成等待结果就可以.  
**应用程序解耦合**:生产者与消费者将不直接耦合,生产者与消费者将依赖MQ进行互相通讯.  
**削峰填谷**:如有并发超过每秒1000的时候时候,就会超到单mysql的阈值,造成数据库宕机,造成整个服务的不可用,如将消息都放到MQ中,根据mysql的阈值进行消费,这样就可以预防mysql宏机,将巅峰时候的并发量积压在MQ中,根据消费能力进行消费,直到积压的消息执行完成,这样就完成填谷.

### 2.AMQP 和 JMS ###

#### AMQP ####

AMQP是一种协议，更准确来说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质区别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。  
AMQP是一个二进制协议，拥有一些现代化特点：多信道、协商式，异步，安全，扩平台，中立，高效。

#### JMS ####

JMS即Java消息服务（JavaMessage Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

#### AMQP和JMS的区别 ####

![在这里插入图片描述][c9629533c5cb444b8aecfff589e601de.png]

#### 目前市面上流行的MQ ####

![在这里插入图片描述][e836c06507d04f808183642ec94665f0.png]

## 三.快速入门-简单入门案例 ##

### 1.添加依赖 ###

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

### 2.编写生成者 ###

public class Producer {
        static final String QUEUE_NAME = "simple_queue";
     
        public static void main(String[] args) throws IOException, TimeoutException {
            //创建连接工厂
            Connection connection = ConnectionUtil.getConnection();
     
            //创建信道
            Channel channel = connection.createChannel();
     
            // 声明（创建）队列      
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
     
            //要发送的信息
            String message = "Hello World";
           
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("已发送消息:"+message);
     
            //关闭资源
            channel.close();
            connection.close();
        }
     
    }

注:在执行上述的消息发送之后；可以登录rabbitMQ的管理控制台

### 3.编写消费者 ###

public class Consumer {
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = ConnectionUtil.getConnection();
     
            //创建频道
            Channel channel = connection.createChannel();
     
            // 声明（创建）队列
            channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null);
     
            //创建消费者:并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //路由key
                    System.out.println("路由key为:"+envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为:"+envelope.getExchange());
                    //消息id
                    System.out.println("消息id为:"+envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("接收到的消息为:"+new String(body));
                }
            };
            //监听消息
            channel.basicConsume(Producer.QUEUE_NAME,true,consumer);
            //不关闭资源，一直监听
        }
    }

上述代码模型  
![在这里插入图片描述][fead8b73cbe44c828ffe4ea33bb2ee6b.png]  
流程解释

![在这里插入图片描述][00f0d81eb24245bd83239c64cd3815b8.png]  
![在这里插入图片描述][240ab3765baf487c94008ca6e8f72c3e.png]

### 4.生产者流转过程 ###

##### 1.客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。 #####

##### 2.客户端调用connection.createChannel方法。此方法开启信道，其包装的channel.open命令发送给Broker,等待channel.basicPublish方法，对应的AMQP命令为Basic.Publish,这个命令包含了content Header 和content Body()。content Header 包含了消息体的属性，例如:投递模式，优先级等，content Body 包含了消息体本身。 #####

##### 3.客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。 #####

### 5.消费者流转过程 ###

##### 1. 消费者客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ，以此通知Broker 本次交互采用的是AMQPO-9-1 协议，紧接着Broker 返回Connection.Start 来建立连接，在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ，Connection.Open/ .Open-Ok 这6 个命令的交互。 #####

##### 2.消费者客户端调用connection.createChannel方法。和生产者客户端一样，协议涉及Channel . Open/Open-Ok命令。 #####

##### 3. 在真正消费之前，消费者客户端需要向Broker 发送Basic.Consume 命令(即调用channel.basicConsume 方法〉将Channel 置为接收模式，之后Broker 回执Basic . Consume - Ok 以告诉消费者客户端准备好消费消息。 #####

##### 4. Broker 向消费者客户端推送(Push) 消息，即Basic.Deliver 命令，这个命令和Basic.Publish 命令一样会携带Content Header 和Content Body。 #####

##### 5.消费者接收到消息并正确消费之后，向Broker 发送确认，即Basic.Ack 命令。 #####

##### 6. 客户端发送完消息需要关闭资源时，涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。 #####

## 四.RabbitMQ工作模式 ##

### 1.Work queues工作队列模式 ###

Work Queues与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。

应用场景：对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。  
![在这里插入图片描述][0bac1b987f2d4fde9cbb78b73459acea.png]

#### 1.1代码示例 ####

生产者

public class Producer {
     
        static final String QUEUE_NAME = "work_queue";
     
        public static void main(String[] args) throws IOException, TimeoutException {
            //本人将获取连接封装成工具类，你们可以把入门案例的连接代码复制过来
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
            // 声明（创建）队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            for (int i = 1; i <= 30; i++) {
                // 发送信息
                String message = "work模式--" + i;
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                System.out.println("已发送消息：" + message);
            }
            // 关闭资源
            channel.close();
            connection.close();
     
        }
    }

消费者

public class Consumer1 {
     
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = ConnectionUtil.getConnection();
            // 创建频道
            final Channel channel = connection.createChannel();
     
            // 声明（创建）队列
            channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
     
            //一次只能接收并处理一个消息
            channel.basicQos(1);
            //创建消费者:并设置消息处理
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    try {
                        //路由key
                        System.out.println("路由key为：" + envelope.getRoutingKey());
                        //交换机
                        System.out.println("交换机为：" + envelope.getExchange());
                        //消息id
                        System.out.println("消息id为：" + envelope.getDeliveryTag());
                        //收到的消息
                        System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
                        Thread.sleep(1000);//单纯增强实验效果
                        //确认信息
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
     
            //监听消息
            channel.basicConsume(Producer.QUEUE_NAME,false,defaultConsumer);
     
        }
    }

测试结果  
![在这里插入图片描述][b6c500cd9814427685b3a029f05fabc9.png]  
![在这里插入图片描述][f4ad31a490754a07b07a0fafc857ad42.png]  
可以从输出的结果看出两台消费者是竞争关系

### 2.Publish/Subscribe 发布订阅模式 ###

###### 模式简介 ######

每个消费者监听自己的队列

生产者将消息发给broker，由交换机将消息转发到绑定此交换机的队列，每个绑定交换机的队列都将接收到消息  
![在这里插入图片描述][b595cb81ded54fbc92e750742b64c3f6.png]  
p(浅蓝):为生产者,不是直接讲消息发送到队列中,而是先将消息发送到x交换机中,由交换机分配消息放到对应的队列中  
x(深蓝): 交换机，不具备存储消息的能力，会将  
C1(C2):消费者，等待处理消息。  
Queue(红色方块):接收并处理消息，不具备储存消息的功能，如果没有队列与其绑定，或者没有符合路由规则的队列，消息就会丢失(gg)…

常见的交换机有如下3种:
    
        ■Fanout:广播型,将消息发送给所有绑定交换机的队列
    
        ■Direct :定向型,把消息发送给指定路由key的队列
    
        ■Topic :通配符，个人理解就是在路由模式(下面会讲到)的队列

#### 代码示例 ####

##### 生产者 #####

/**
     * 发布与订阅使用的交换机类型为：fanout
     */
    public class Producer {
     
        //交换机名称
        static final String FANOUT_EXCHAGE = "fanout_exchange";
        //队列名称
        static final String FANOUT_QUEUE_1 = "fanout_queue_1";
        //队列名称
        static final String FANOUT_QUEUE_2 = "fanout_queue_2";
     
        public static void main(String[] args) throws IOException, TimeoutException {
            //创建连接
            Connection connection = ConnectionUtil.getConnection();
     
            // 创建信道
            Channel channel = connection.createChannel();
            
            channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
            //创建队列
            channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);
            channel.queueDeclare(FANOUT_QUEUE_2,true,false,false,null);
     
            //队列绑定交换机
            channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");
            channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"");
     
            for (int i = 1; i <= 10; i++) {
                // 发送信息
                String message = "岳某人 发布订阅模式--" + i;
                channel.basicPublish(FANOUT_EXCHAGE,"",null,message.getBytes());
                System.out.println("已发送信息:"+message);
            }
            // 关闭资源
            channel.close();
            connection.close();
        }
    }

##### 消费者 #####

public class Consumer1 {
     
        public static void main(String[] args) throws Exception {
            Connection connection = ConnectionUtil.getConnection();
     
            // 创建频道
            Channel channel = connection.createChannel();
     
            //声明交换机
            channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);
     
            // 声明（创建）队列
            channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null);
     
            //队列绑定交换机
            channel.queueBind(Producer.FANOUT_QUEUE_1, Producer.FANOUT_EXCHAGE, "");
     
            //创建消费者；并设置消息处理
            DefaultConsumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,         AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
                }
            };
            //监听消息
            channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer);
        }
    }

##### 测试结构 #####

![在这里插入图片描述][824d25514ac645f6bba4c7fea4fd41cd.png]  
![在这里插入图片描述][0236f84123264b44884ed7967575c9e6.png]

### 3.Routing路由模式 ###

#### 模式介绍 ####

##### 发送方向交换机发送消息时，必须指定消息的RoutingKey #####

##### 交换机不再把消息交给每一个绑定的队列，而是根据消息的RoutingKey判断筛选，只有队列的RoutingKey与消息的RoutingKey一致，才会接收到消息 #####

![在这里插入图片描述][be6d7244b0594fd58d99f682220dea03.png]

#### ■生产者 ####

/**
     * 路由模式的交换机类型为：direct
     */
    public class Producer {
        //交换机名称
        static final String DIRECT_EXCHAGE = "direct_exchange";
        //队列名称
        static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
        //队列名称
        static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";
     
        public static void main(String[] args) throws IOException, TimeoutException {
            //创建连接
            Connection connection = ConnectionUtil.getConnection();
     
            // 创建频道
            Channel channel = connection.createChannel();
     
            //声明交换机
            channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
     
            //声明队列
            channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
            channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);
     
            //与队列绑定交换机
            channel.queueBind(DIRECT_QUEUE_INSERT,DIRECT_EXCHAGE,"insert");
            channel.queueBind(DIRECT_QUEUE_UPDATE,DIRECT_EXCHAGE,"update");
            //发送信息
            String message = "cxk 路由模式；routing key 为 insert";
            channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes());
            System.out.println("已发送信息:" + message);
     
            //发送信息
            message = "Chicken you are too beautiful 路由模式；routing key 为 update";
            channel.basicPublish(DIRECT_EXCHAGE,"update",null,message.getBytes());
            System.out.println("已发送信息:" + message);
            // 关闭资源
            channel.close();
            connection.close();
        }
    }

#### ■消费者 ####

public class Consumer1 {
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = ConnectionUtil.getConnection();
     
            // 创建频道
            Channel channel = connection.createChannel();
     
            //声明交换机
            channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT);
     
            //声明队列
            channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null);
     
            //队列绑定交换机
            channel.queueBind(Producer.DIRECT_QUEUE_INSERT,Producer.DIRECT_EXCHAGE,"insert");
     
            //创建消费者，并设置消息处理
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
                }
            };
     
            //监听消息
            channel.basicConsume(Producer.DIRECT_QUEUE_INSERT,true,defaultConsumer);
        }
    }

测试结果  
1.消费者  
![在这里插入图片描述][13e4ff7450474273834460cd44b8f94f.png]  
2.消费者  
![在这里插入图片描述][4752ea1c7d5a44c0b737931b6aa2201a.png]

### 4.Topics通配符模式 ###

#### 模式介绍 ####

个人理解：和Routing模式很类似，只不过Routing模式的路由键是固定的，Topic类型的交换机可以模糊匹配路由键(非固定)

开发中路由键一般都是多个单词拼凑的，中间用.连接，例如:ys.add
    
        通配符规则:

‘\#’:匹配一个或多个词

‘\*’:只能匹配一个词

#### 代码示例 ####

##### ■生产者 #####

public class Producer {
        //交换机名称
        static final String TOPIC_EXCHAGE = "topic_exchange";
        //队列名称
        static final String TOPIC_QUEUE_1 = "topic_queue_1";
        //队列名称
        static final String TOPIC_QUEUE_2 = "topic_queue_2";
     
        public static void main(String[] args) throws IOException, TimeoutException {
            //创建连接
            Connection connection = ConnectionUtil.getConnection();
     
            // 创建频道
            Channel channel = connection.createChannel();
            //声明交换机
            channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
     
            //发送消息
            String message = "新增了商品。Topic模式；routing key 为 item.insert " ;
            channel.basicPublish(TOPIC_EXCHAGE,"item.insert",null,message.getBytes());
            // 发送信息
            message = "修改了商品。Topic模式；routing key 为 item.update" ;
            channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes());
            System.out.println("已发送消息：" + message);
     
            // 发送信息
            message = "删除了商品。Topic模式；routing key 为 item.delete" ;
            channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes());
            System.out.println("已发送消息：" + message);
     
            // 关闭资源
            channel.close();
            connection.close();
        }
    }

##### ■消费者 #####

public class Consumer1 {
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = ConnectionUtil.getConnection();
     
            // 创建频道
            Channel channel = connection.createChannel();
            //声明交换机
            channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
     
            //声明队列
            channel.queueDeclare(Producer.TOPIC_QUEUE_1,true,false,false,null);
     
            //队列绑定交换机
            channel.queueBind(Producer.TOPIC_QUEUE_1,Producer.TOPIC_EXCHAGE,"item.update");
            channel.queueBind(Producer.TOPIC_QUEUE_1,Producer.TOPIC_EXCHAGE,"item.delete");
     
            //创建消费者
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //路由key
                    System.out.println("路由key为：" + envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为：" + envelope.getExchange());
                    //消息id
                    System.out.println("消息id为：" + envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));
                }
            };
            //监听消息
            channel.basicConsume(Producer.TOPIC_QUEUE_1,true,defaultConsumer);
        }
    }

测试结构  
1.消费者1  
![在这里插入图片描述][8299cb2091dd4bd69971713f1d2b7eca.png]  
2.消费者2  
![在这里插入图片描述][4d1d24a74f3648f2a337dab091f164f6.png]

![在这里插入图片描述][ea7d2ba0b3cc4992aa70b8ae62a696e5.png]

--------------------

[bdc6632f6af346f09f52c2c251698181.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/50f5d03fd6ce49da8d8af008a807eb29.png
[c9629533c5cb444b8aecfff589e601de.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/a75885ffb56e48e5a638da1bcbcdb241.png
[e836c06507d04f808183642ec94665f0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/2af65a9a504c4db29d1a321b756c9171.png
[fead8b73cbe44c828ffe4ea33bb2ee6b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/9e3e2e805c26413d9048cc8344401751.png
[00f0d81eb24245bd83239c64cd3815b8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/4bafbfbee24847949ce304b9dde93d72.png
[240ab3765baf487c94008ca6e8f72c3e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/bbe49e4bd19345a0b38d77b1a76522fe.png
[0bac1b987f2d4fde9cbb78b73459acea.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/6bfe599b591649dea435ebfdbeabe21e.png
[b6c500cd9814427685b3a029f05fabc9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/dae2e572225d405eb3d0b1a3513fdc94.png
[f4ad31a490754a07b07a0fafc857ad42.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/35354450ad934982ba5e4f92b246ce78.png
[b595cb81ded54fbc92e750742b64c3f6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/1cbcb07cdbdb4b23b9b821f02e4233c6.png
[824d25514ac645f6bba4c7fea4fd41cd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/74946b67fb5c4eb8bfc87ec31bb96e7b.png
[0236f84123264b44884ed7967575c9e6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/c80f162538344b44b0f45eb83c9fcc60.png
[be6d7244b0594fd58d99f682220dea03.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/7dd50654a0064627abffad824e1819a3.png
[13e4ff7450474273834460cd44b8f94f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/700fc4590e7c4586b501900e6f2a59ce.png
[4752ea1c7d5a44c0b737931b6aa2201a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/3a6a55d5fcdd4a9d9d289ed6c05f1abd.png
[8299cb2091dd4bd69971713f1d2b7eca.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/a7e580a2249545b0a8370d3a8ce16db5.png
[4d1d24a74f3648f2a337dab091f164f6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/64c6e9d096ea4382983ad3da9c05ec9c.png
[ea7d2ba0b3cc4992aa70b8ae62a696e5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/7b4e95fb4778407fbd96cd6b0ff0f0fe.png