RabbitMQ的使用及其支持的消息类型

我就是我 2023-03-05 11:56 6阅读 0赞

### RabbitMQ的使用及其支持的消息类型 ###

*   *  1、RabbitMQ的初使用
     *   *  1.1、创建一个用户
         *  1.2、创建虚拟主机
         *  1.3、绑定虚拟主机和用户
     *  2、RabbitMQ支持的消息模式
     *   *  2.1、AMQP协议的回顾
         *  2.2、官方RabbitMQ所支持的消息模式
     *  2.3、简单模式(HelloWorld)
     *   *  2.3.1、名词解释
         *  2.3.2、代码实现
     *  2.4、任务模式(WorkQueues)
     *   *  2.4.1、名词解释
         *  2.4.2、代码实现
         *  2.4.3、针对2.4.2产生的三种问题，设置手动确认机制、队列消息持久化、一次只消费一条消息的代码实现
     *  2.5、发布订阅模式(Publish/Subscribe)
     *   *  2.5.1、名词解释
         *  2.5.2、代码实现
     *  2.6、路由模式(Routing)
     *   *  2.6.1、名词解释
         *  2.6.2、代码实现
     *  2.7、主题模式(Topic)
     *   *  2.7.1、名词解释
         *  2.7.2、代码实现
     *  2.8、代码GitHub地址：\[https://github.com/Amywang1996/RabbitMQ.git\](https://github.com/Amywang1996/RabbitMQ.git)

在使用RabbitMQ时，我们会先创建一个用户，然后创建一个虚拟主机，给用户添加权限。

## 1、RabbitMQ的初使用 ##

### 1.1、创建一个用户 ###

![在这里插入图片描述][20200804174929346.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70]  
上面的Tags选项，其实是指定用户的角色，可选的有以下几个：

*  **超级管理员(administrator)**  
    可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。
 *  **监控者(monitoring)**  
    可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)
 *  **策略制定者(policymaker)**  
    可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息。
 *  **普通管理者(management)**  
    仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。
 *  **其他**  
    无法登陆管理控制台，通常就是普通的生产者和消费者。

### 1.2、创建虚拟主机 ###

为了让各个用户可以互不干扰的工作，RabbitMQ添加了虚拟主机（Virtual Hosts）的概念。其实就是一个独立的访问路径，不同用户使用不同路径，各自有自己的队列、交换机，互相不会影响。  
![在这里插入图片描述][20200804185126573.png]

### 1.3、绑定虚拟主机和用户 ###

创建好虚拟主机，我们还要给用户添加访问权限：  
![在这里插入图片描述][202008041853081.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][20200804185336871.png]

## 2、RabbitMQ支持的消息模式 ##

### 2.1、AMQP协议的回顾 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 2]

### 2.2、官方RabbitMQ所支持的消息模式 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 5]  
RabbitMQ支持七种消息模式，本篇只总结前五种，之后会持续更新。

## 2.3、简单模式(HelloWorld) ##

### 2.3.1、名词解释 ###

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

*  P：生产者，也就是要发送消息的程序。
 *  C：消费者：消息的接受者，会一直等待消息到来。
 *  queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

### 2.3.2、代码实现 ###

**1、与RabbitMQ服务器建立连接工具类**

/**
     * RabbitMQ连接关闭工具类
     */
    public class RabbitMQUtils {
        
       private static ConnectionFactory connectionFactory;
       private static Properties properties;
       static{
        
           //重量级资源  类加载只执行一次
           connectionFactory = new ConnectionFactory();
           connectionFactory.setHost("192.168.46.102");
           connectionFactory.setPort(5672);
           connectionFactory.setVirtualHost("/study");
           connectionFactory.setUsername("Amy");
           connectionFactory.setPassword("Amy");
       }
       //定义提供连接对象的方法
       public static Connection getConnection() {
        
           try {
        
               return connectionFactory.newConnection();
           } catch (Exception e) {
        
               e.printStackTrace();
           }
           return null;
       }
       //关闭通道和关闭连接工具方法
       public static void closeConnectionAndChanel(Channel channel, Connection conn) {
        
           try {
        
               if(channel!=null) channel.close();
               if(conn!=null)   conn.close();
           } catch (Exception e) {
        
               e.printStackTrace();
    
           }
       }
    }

**2、生产者发送消息**

/**
     * 简单模式生产者
     */
    public class Provider {
        
    //生产消息
     @Test
     public void testSendMessage() throws IOException, TimeoutException {
        
         //通过工具类获取连接对象
         Connection connection = RabbitMQUtils.getConnection();
         //获取连接中通道
         Channel channel = connection.createChannel();
         //通道绑定对应消息队列
         //参数1:  队列名称 如果队列不存在自动创建
         //参数2:  用来定义队列特性是否要持久化 true 持久化队列   false 不持久化
         //参数3:  exclusive 是否独占队列  true 独占队列   false  不独占
         //参数4:  autoDelete: 是否在消费完成后自动删除队列  true 自动删除  false 不自动删除
         //参数5:  额外附加参数
         channel.queueDeclare("hello",false,false,false,null);
         //发布消息
         //参数1: 交换机名称 参数2:队列名称  参数3:传递消息额外设置(MessageProperties.PERSISTENT_TEXT_PLAIN表示消息的持久化)  参数4:消息的具体内容
         channel.basicPublish("","hello",null,"hello rabbitmq".getBytes());
         //调用工具类，关闭连接
         RabbitMQUtils.closeConnectionAndChanel(channel,connection);
     }
    }

**3、消费者接收消息并做处理**

/**
     * 简单模式消费者
     */
    public class Customer {
        
    public static void main(String[] args) throws IOException, TimeoutException {
        
          //通过工具类获取连接
          Connection connection = RabbitMQUtils.getConnection();
          //创建通道
          Channel channel = connection.createChannel();
          //通道绑定对象
          channel.queueDeclare("hello",false,false,false,null);
          //消费消息
          //参数1: 消费那个队列的消息 队列名称
          //参数2: 开始消息的自动确认机制
          //参数3: 消费时的回调接口
          channel.basicConsume("hello",true,new DefaultConsumer(channel){
        
              @Override //最后一个参数: 消息队列中取出的消息
              public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                  System.out.println("===================================="+new String(body));
              }
          });
      }
    }

**4、发送接收操作queues的概况图**  
生产者发送消息消费者未接收：  
![在这里插入图片描述][20200804203748223.png]  
生产者发送的消息已被消费者接收：  
![在这里插入图片描述][20200804203905899.png]  
**特别注意**：  
channel.queueDeclare(“hello”,true,false,false,null);  
声明队列的五个参数必须要一致，否则会造成生产者发送的消息，消费者接收不到的情况。

## 2.4、任务模式(WorkQueues) ##

消息处理比较耗时的时候，可能生产消息的速度会远远大于消费消息的速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work 模型：**让多个消费者绑定到一个队列，共同消费队列中的消息**。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。

### 2.4.1、名词解释 ###

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

### 2.4.2、代码实现 ###

**1、与RabbitMQ服务器建立连接工具类**  
同简单模式  
**2、生产者发送消息**  
实现代码生产者发送20条消息

public class Provider1 {
        
     public static void main(String[] args) throws IOException {
        
         //获取连接对象
         Connection connection = RabbitMQUtils.getConnection();
         //获取通道对象
         Channel channel = connection.createChannel();
         //通过通道声明队列:队列名称：work1，队列不持久化，不独占队列，队列消费完后不自动删除，无额外参数
         channel.queueDeclare("work1", false, false, false, null);
         for (int i = 1; i <=20; i++) {
        
             //生产消息：无交换机，队列名称：work1,无传递消息额外设置，传递消息内容
             channel.basicPublish("", "work1", null, (i + "hello work queues").getBytes());
         }
         //关闭资源
         RabbitMQUtils.closeConnectionAndChanel(channel, connection);
     }
    }

**默认情况下，RabbitMQ将每个消息按顺序发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。**  
**2、默认情况下循环公平的发送消息，消费者1消费消息**

public class FairCustomer1 {
        
     public static void main(String[] args) throws IOException {
        
        //获取连接
         Connection connection = RabbitMQUtils.getConnection();
         final Channel channel = connection.createChannel();
         //通过通道声明队列:队列名称：work1，队列不持久化，不独占队列，队列消费完后不自动删除，无额外参数
         channel.queueDeclare("work1",false,false,false,null);
         //参数1:队列名称 ，参数2:消息自动确认 true消费者自动向rabbitmq确认消息消费 ，false 不会自动确认消息
         channel.basicConsume("work1",true,new DefaultConsumer(channel){
        
             @Override //最后一个参数: 消息队列中取出的消息
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                 try{
        
                     Thread.sleep(2000);
                 }catch (Exception e){
        
                     e.printStackTrace();
                 }
                 System.out.println("消费者-1: "+new String(body));
             }
         });
      }
    }

**3、默认情况下循环公平的发送消息，消费者2消费消息**

public class FairCustomer2 {
        
     public static void main(String[] args) throws IOException {
        
        //获取连接
         Connection connection = RabbitMQUtils.getConnection();
         final Channel channel = connection.createChannel();
         //通过通道声明队列:队列名称：work1，队列不持久化，不独占队列，队列消费完后不自动删除，无额外参数
         channel.queueDeclare("work1",false,false,false,null);
         //参数1:队列名称 ，参数2:消息自动确认 true消费者自动向rabbitmq确认消息消费 ，false 不会自动确认消息
         channel.basicConsume("work1",true,new DefaultConsumer(channel){
        
             @Override //最后一个参数: 消息队列中取出的消息
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                 try{
        
                     Thread.sleep(2000);
                 }catch (Exception e){
        
                     e.printStackTrace();
                 }
                 System.out.println("消费者-1: "+new String(body));
             }
         });
      }
    }

**4、默认情况下循环发送，公平接收消息输出结果**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 7]  
上述默认的处理方式可能会出现一些问题：  
**1、多个消费者，有一个消费者死亡，消息丢失问题：**  
完成一项任务可能需要几秒钟。如果消费者FairCustomer1开始一项漫长的任务，而仅部分完成而死掉，会发生什么情况呢?RabbitMQ一旦向消费者发送了一条消息，便立即将其标记为删除。在这种情况下，如果消费者FairCustomer1死掉，我们将丢失正在处理的消息。我们还将丢失所有发送给FairCustomer1但尚未处理的消息。但是我们不想丢失任何任务。如果FairCustomer1死亡，我们希望将任务交付给FairCustomer2。为了确保消息永不丢失，RabbitMQ支持 消息确认。消费者发送回确认，以告知RabbitMQ已经接收，处理了特定的消息，并且RabbitMQ可以自由删除它。如果消费者在不发送确认的情况下死亡（其通道已关闭，连接已关闭或TCP连接丢失），RabbitMQ将了解消息未得到充分处理，并将重新排队。如果同时有其他消费者在线，它将很快将其重新分发给另一个消费者。这样，可以确保即使某一个消费者偶尔死亡也不会丢失任何消息。没有任何消息超时；消费者死亡时，RabbitMQ将重新传递消息。即使处理消息需要非常非常长的时间也没关系。默认情况下，手动消息确认处于打开状态。在前面的示例中，我们通过autoAck = true 标志显式关闭了它们。现在，是时候将该标志设置为false并在工作完成后从消费者发送适当的确认。  
**2、队列、消息未持久化**  
我们已经学会了如何确保即使消费者死亡，任务也不会丢失，即设置消息手动确认。但是，如果RabbitMQ服务器停止，我们的任务仍然会丢失。RabbitMQ退出或崩溃时，除非告诉它不要这样做，否则它将忘记队列和消息。要确保消息不会丢失，需要做两件事：我们需要将队列和消息都标记为持久性。 声明队列持久化：channel.queueDeclare()方法的第二个参数设置为true,以确保即使RabbitMQ重新启动，队列也不会丢失；消息标记为持久化： channel.basicPublish()的第二个参数设置为：MessageProperties.PERSISTENT\_TEXT\_PLAIN。  
**3、多个消费者同时执行任务平均分配，任务执行所需时间不同造成有的消费者空闲，有的消费者忙碌的状况**  
发生这种情况是因为RabbitMQ在消息进入队列时才调度消息。它不会查看消费者的未确认消息数。它只是盲目地将每第n条消息发送给第n个消费者。  
为了解决这个问题，我们可以将basicQos方法与 prefetchCount = 1设置一起使用。这告诉RabbitMQ一次不要给消费者一个以上的消息。换句话说，在处理并确认上一条消息之前，不要将新消息发送给消费者。而是将其分派给不忙的下一个工作程序。

### 2.4.3、针对2.4.2产生的三种问题，设置手动确认机制、队列消息持久化、一次只消费一条消息的代码实现 ###

**1、与RabbitMQ服务器建立连接工具类**  
同简单模式  
**2、生产者发送消息**  
实现代码生产者发送20条消息

public class Provider2 {
        
        public static void main(String[] args) throws IOException {
        
           //获取连接对象
           Connection connection = RabbitMQUtils.getConnection();
           //获取通道对象
           Channel channel = connection.createChannel();
           //通过通道声明队列:队列名称：work1，队列不持久化，不独占队列，队列消费完后不自动删除，无额外参数
           //设置队列出持久化
           channel.queueDeclare("work2", true, false, false, null);
           for (int i = 1; i <=20; i++) {
        
               //生产消息：无交换机，队列名称：work1,无传递消息额外设置，传递消息内容
               //设置消息持久化:MessageProperties.PERSISTENT_TEXT_PLAIN
               channel.basicPublish("", "work2", MessageProperties.PERSISTENT_TEXT_PLAIN, (i + "hello work queues").getBytes());
           }
           //关闭资源
           RabbitMQUtils.closeConnectionAndChanel(channel, connection);
        }
    }

**3、消费者1处理4秒钟确认收到消息**

public class Customer1 {
        
       public static void main(String[] args) throws IOException {
        
           //获取连接
           Connection connection = RabbitMQUtils.getConnection();
           final Channel channel = connection.createChannel();
           channel.basicQos(1);//一次只接受一条未确认的消息
           //通过通道声明队列:队列名称：work1，队列持久化，不独占队列，队列消费完后不自动删除，无额外参数
           channel.queueDeclare("work2", true, false, false, null);
           //参数1:队列名称 ，参数2:消息自动确认 true消费者自动向rabbitmq确认消息消费 ，false 不会自动确认消息
           //关闭自动确认，启动手动确认消息
           channel.basicConsume("work2", false, new DefaultConsumer(channel) {
        
               @Override //最后一个参数: 消息队列中取出的消息
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                   try {
        
                       Thread.sleep(4000);
                   } catch (Exception e) {
        
                       e.printStackTrace();
                   }
                   System.out.println("消费者-1: " + new String(body));
                   //手动确认  参数1:手动确认消息标识  参数2:false 每次确认一个
                   channel.basicAck(envelope.getDeliveryTag(), false);
               }
           });
       }
    }

**4、消费者2处理2秒钟确认收到消息**

public class Customer2 {
        
       public static void main(String[] args) throws IOException {
        
           //获取连接
           Connection connection = RabbitMQUtils.getConnection();
           final Channel channel = connection.createChannel();
           channel.basicQos(1);//一次只接受一条未确认的消息
           //通过通道声明队列:队列名称：work1，队列持久化，不独占队列，队列消费完后不自动删除，无额外参数
           channel.queueDeclare("work2", true, false, false, null);
           //参数1:队列名称 ，参数2:消息自动确认 true消费者自动向rabbitmq确认消息消费 ，false 不会自动确认消息
           //关闭自动确认，启动手动确认消息
           channel.basicConsume("work2", false, new DefaultConsumer(channel) {
        
               @Override //最后一个参数: 消息队列中取出的消息
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                   try {
        
                       Thread.sleep(2000);
                   } catch (Exception e) {
        
                       e.printStackTrace();
                   }
                   System.out.println("消费者-2: " + new String(body));
                   //手动确认  参数1:手动确认消息标识  参数2:false 每次确认一个
                   channel.basicAck(envelope.getDeliveryTag(), false);
               }
           });
       }
    }

**5、消费者1消费者2消费消息输出结果**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 8]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 9]

## 2.5、发布订阅模式(Publish/Subscribe) ##

### 2.5.1、名词解释 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70_pic_center]

*  P：生产者，也就是要发送消息的程序。
 *  C：消费者：消息的接受者，会一直等待消息到来。
 *  queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。
 *  X：交换机

### 2.5.2、代码实现 ###

在广播模式下，消息发送流程是这样的：

*  可以有多个消费者
 *  每个**消费者有自己的queue**（队列）
 *  每个**队列都要绑定到Exchange**（交换机）
 *  **生产者发送的消息，只能发送到交换机**，交换机来决定要发给哪个队列，生产者无法决定。
 *  交换机把消息发送给绑定过的所有队列
 *  队列的消费者都能拿到消息。实现一条消息被多个消费者消费  
    **1、与RabbitMQ服务器建立连接工具类**  
    同简单模式  
    **2、生产者发送消息**

public class Provider {
        
       public static void main(String[] args) throws IOException {
        
           //获取连接对象
           Connection connection = RabbitMQUtils.getConnection();
           Channel channel = connection.createChannel();
           //将通道声明指定交换机   //参数1: 交换机名称    参数2: 交换机类型  fanout 广播类型
           channel.exchangeDeclare("logs","fanout");
           //发送消息
           channel.basicPublish("logs","",null,"fanout type message".getBytes());
           //释放资源
           RabbitMQUtils.closeConnectionAndChanel(channel,connection);
       }
    }

**3、消费者1**

public class Customer1 {
        
       public static void main(String[] args) throws IOException {
        
           //获取连接对象
           Connection connection = RabbitMQUtils.getConnection();
           Channel channel = connection.createChannel();
           //通道绑定交换机
           channel.exchangeDeclare("logs", "fanout");
           //临时队列
           String queueName = channel.queueDeclare().getQueue();
           //绑定交换机和队列
           channel.queueBind(queueName, "logs", "");
           //消费消息
           channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
        
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                   System.out.println("消费者1: " + new String(body));
               }
           });
       }
    }

**4、消费者2**

public class Customer2 {
        
      public static void main(String[] args) throws IOException {
        
           //获取连接对象
           Connection connection = RabbitMQUtils.getConnection();
           Channel channel = connection.createChannel();
           //通道绑定交换机
           channel.exchangeDeclare("logs", "fanout");
           //临时队列
           String queueName = channel.queueDeclare().getQueue();
           //绑定交换机和队列
           channel.queueBind(queueName, "logs", "");
           //消费消息
           channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
        
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                   System.out.println("消费者2: " + new String(body));
               }
           });
       }
    }

**5、消费者3**

public class Customer3 {
        
     public static void main(String[] args) throws IOException {
        
          //获取连接对象
          Connection connection = RabbitMQUtils.getConnection();
          Channel channel = connection.createChannel();
          //通道绑定交换机
          channel.exchangeDeclare("logs","fanout");
          //临时队列
          String queueName = channel.queueDeclare().getQueue();
          //绑定交换机和队列
          channel.queueBind(queueName,"logs","");
          //消费消息
          channel.basicConsume(queueName,true,new DefaultConsumer(channel){
        
              @Override
              public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                  System.out.println("消费者3: "+new String(body));
              }
          });
      }
    }

**6、消费者1、2、3消费消息输出结果**  
![在这里插入图片描述][20200804231907305.png]  
![在这里插入图片描述][20200804231924537.png]  
![在这里插入图片描述][20200804231948158.png]

## 2.6、路由模式(Routing) ##

### 2.6.1、名词解释 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 10]

*  P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
 *  X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
 *  C1：消费者，其所在队列指定了需要routing key 为 error 的消息
 *  C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

### 2.6.2、代码实现 ###

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。  
在Direct模型下：

*  队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
 *  消息的发送方在向 Exchange发送消息时，也必须指定消息的 RoutingKey。
 *  Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息  
    **1、与RabbitMQ服务器建立连接工具类**  
    同简单模式  
    **2、生产者发送消息**

public class Provider {
        
      public static void main(String[] args) throws IOException {
        
           //获取连接对象
           Connection connection = RabbitMQUtils.getConnection();
           //获取连接通道对象
           Channel channel = connection.createChannel();
           String exchangeName = "logs_direct";
           //通过通道声明交换机  参数1:交换机名称  参数2:direct  路由模式
           channel.exchangeDeclare(exchangeName,"direct");
           //发送消息
           String routingkey = "info";
           channel.basicPublish(exchangeName,routingkey,null,("这是direct模型发布的基于route key: ["+routingkey+"] 发送的消息").getBytes());
           //关闭资源
           RabbitMQUtils.closeConnectionAndChanel(channel,connection);
       }
    }

**3、消费者1**

public class Customer1 {
        
       public static void main(String[] args) throws IOException {
        
           Connection connection = RabbitMQUtils.getConnection();
           Channel channel = connection.createChannel();
           String exchangeName = "logs_direct";
           //通道声明交换机以及交换的类型
           channel.exchangeDeclare(exchangeName,"direct");
           //创建一个临时队列
           String queue = channel.queueDeclare().getQueue();
           //基于route key绑定队列和交换机
           channel.queueBind(queue,exchangeName,"error");
           //获取消费的消息
           channel.basicConsume(queue,true,new DefaultConsumer(channel){
        
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                   System.out.println("消费者1: "+ new String(body));
               }
           });
       }
    }

**4、消费者2**

public class Customer2 {
        
      public static void main(String[] args) throws IOException {
        
           Connection connection = RabbitMQUtils.getConnection();
           Channel channel = connection.createChannel();
           String exchangeName = "logs_direct";
           //声明交换机 以及交换机类型 direct
           channel.exchangeDeclare(exchangeName, "direct");
           //创建一个临时队列
           String queue = channel.queueDeclare().getQueue();
           //临时队列和交换机绑定
           channel.queueBind(queue, exchangeName, "info");
           channel.queueBind(queue, exchangeName, "error");
           channel.queueBind(queue, exchangeName, "warning");
           //消费消息
           channel.basicConsume(queue, true, new DefaultConsumer(channel) {
        
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                   System.out.println("消费者2: " + new String(body));
               }
           });
       }
    }

## 2.7、主题模式(Topic) ##

### 2.7.1、名词解释 ###

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

*  P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
 *  X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
 *  C1：消费者，其所在队列指定了需要routing key 包含orange的消息
 *  C2：消费者，其所在队列指定了需要routing key 包含rabbit结尾和lazy后有一个次的消息

### 2.7.2、代码实现 ###

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！这种模型Routingkey 一般都是由一个或多个单词组成，多个单词之间以”.”分割，例如：item.insert  
\*(star) can substitute for exactly one word. 匹配不多不少恰好1个词  
\#(hash) can substitute for zero or more words. 匹配一个或多个词  
**1、与RabbitMQ服务器建立连接工具类**  
同简单模式  
**2、生产者发送消息**

public class Provider {
        
       public static void main(String[] args) throws IOException {
        
           //获取连接对象
           Connection connection = RabbitMQUtils.getConnection();
           Channel channel = connection.createChannel();
           //声明交换机以及交换机类型 topic
           channel.exchangeDeclare("topics", "topic");
           //发布消息
           String routekey = "save.user.delete";
           channel.basicPublish("topics", routekey, null, ("这里是topic动态路由模型,routekey: [" + routekey + "]").getBytes());
           //关闭资源
           RabbitMQUtils.closeConnectionAndChanel(channel, connection);
    
       }
    }

**3、消费者1消费消息**

public class Customer1 {
        
       public static void main(String[] args) throws IOException {
        
           //获取连接
           Connection connection = RabbitMQUtils.getConnection();
           Channel channel = connection.createChannel();
           //声明交换机以及交换机类型
           channel.exchangeDeclare("topics", "topic");
           //创建一个临时队列
           String queue = channel.queueDeclare().getQueue();
           //绑定队列和交换机  动态统配符形式route key
           channel.queueBind(queue, "topics", "*.user.*");
           //消费消息
           channel.basicConsume(queue, true, new DefaultConsumer(channel) {
        
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                   System.out.println("消费者1: " + new String(body));
               }
           });
       }
    }

**4、消费者2消费消息**

public class Customer2 {
        
       public static void main(String[] args) throws IOException {
        
           //获取连接
           Connection connection = RabbitMQUtils.getConnection();
           Channel channel = connection.createChannel();
           //声明交换机以及交换机类型
           channel.exchangeDeclare("topics", "topic");
           //创建一个临时队列
           String queue = channel.queueDeclare().getQueue();
           //绑定队列和交换机  动态统配符形式route key
           channel.queueBind(queue, "topics", "*.user.#");
           //消费消息
           channel.basicConsume(queue, true, new DefaultConsumer(channel) {
        
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                   System.out.println("消费者2: " + new String(body));
               }
           });
       }
    }

## 2.8、代码GitHub地址：[https://github.com/Amywang1996/RabbitMQ.git][https_github.com_Amywang1996_RabbitMQ.git] ##

[20200804174929346.png]: /images/20230208/6ee1ba3c9b9e4c88889408358e754eb6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70]: /images/20230208/feb6d9097f2340f2937acbeb2d146c62.png
[20200804185126573.png]: /images/20230208/8a57a082b2034a3bbff9186376955b23.png
[202008041853081.png]: /images/20230208/0a742b28983343778c14e45ac6f7c568.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 1]: /images/20230208/8f8e56dedc594c97b1c3fa38520a4bfa.png
[20200804185336871.png]: /images/20230208/9037590112a84ffbbd10fde1dfc3f2b4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 2]: /images/20230208/f622525b70e947f1802a1c6c9a27f2b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 3]: /images/20230208/227d5cf86a504f85947958284ed2dc6c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 4]: /images/20230208/a64037e7c7574a42b3fcad0d40a1e8f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 5]: /images/20230208/014773ce8d8e454598af4188bef4c441.png
[20200804192030621.png]: /images/20230208/2fc01bf0aed24b508093a60287556028.png
[20200804203748223.png]: /images/20230208/a48294e4b2054ec697154cb2caaacf0b.png
[20200804203905899.png]: /images/20230208/5149b8435e554c908d1100758fffdada.png
[20200804205129172.png]: /images/20230208/07c5edc34e3544f5895db3b137f51690.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 6]: /images/20230208/cbe3adaad95d4af293097f7ee0e5d090.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 7]: /images/20230208/3499c60fd8ed42019ecb90582aa1a3c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20200804225254277.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 9]: /images/20230208/f6762cbcfa924199a36a85933e369cf2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70_pic_center]: /images/20230208/4a9deb7786a44e6eaaa0001583fede86.png
[20200804231907305.png]: /images/20230208/ecdaefe088d2470cb87a2522b7eda3cc.png
[20200804231924537.png]: /images/20230208/3f2c1314029b49ab9c5e6426d5807364.png
[20200804231948158.png]: /images/20230208/6eaa1c224748425b8f8765a1eb03f9bc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0ODk2NzMw_size_16_color_FFFFFF_t_70 10]: /images/20230208/fa887c5a4da748f0b7860f6360ca7bde.png
[20200804234803305.png]: /images/20230208/dfd25d2f06a0481e992f42350b78545e.png
[https_github.com_Amywang1996_RabbitMQ.git]: https://github.com/Amywang1996/RabbitMQ.git