• RabbitMQ的入门篇

    目录

    1. RabbitMQ简介

      1.1 RabbitMQ

      1.2 RabbitMQ的介绍

      1.3 RabbitMQ的特点

    2. RabbitMQ的安装

      2.1 RabbitMQ下载 

      2.2 下载的安装包

    ​  2.3  安装步骤

      2.4  登录访问

     3. RabbitMQ的入门

      3.1 RabbitMQ的架构原理

      3.2 永远的hello world程序 

    4. RabbitMQ的工作模式

     4.1 简单模式(Hello World)

     4.2 工作队列模式(Work Queues) 

     4.3 发布订阅模式(Publish/Subscribe) 

     4.4 路由模式(Routing) 

     4.5 主题模式(Topics) 

     4.6 发布确认模式(Publisher Confirms)  

    5. RabbitMQ的注解开发

     5.1 简单模式

     5.2 工作队列

     5.3 发布订阅

     5.4 路由模式

     5.5 主题模式

     5.6 发布确认 

         学习一门新技术，我们总会遇到许许多多的陌生名词，我们可以从我们熟悉的点切入，去学习和关联这门我们不熟悉的技术。RabbitMQ我们比较熟悉的是Q(Queue)队列的意思，那RabbitMQ是不是就是一个保存消息的队列，更准确的说法是，RabbitMQ是一个消息代理中间件。

    1. RabbitMQ简介

       1.1 RabbitMQ

      基于AMQP协议，erlang语言开发，是部署最广泛的开源消息中间件，是最受欢迎的开源消息中间件之一

       1.2 RabbitMQ的介绍

            RabbitMQ是使用Erlang语言来编写的，实现了AMQP协议的一款MQ产品。AMQP，即Advanced Message Queuing Protocol(高级消息队列协议)，是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。MQ(Message Queue)消息队列是以生产者和消费者的异步通信原理来实现系统解耦合的。

       1.3 RabbitMQ的特点

    • 可靠性。支持持久化，传输确认，发布确认，消费端消息手动确认等保证了MQ的可靠性 。
    • 灵活的分发消息策略。分发消息策略有：简单模式、工作队列模式、发布订阅模式、路由模式、主题模式。
    • 支持集群。多台RabbitMQ服务器可以组成一个集群，形成一个逻辑Broker。
    • 多种协议。RabbitMQ支持多种消息队列协议，比如 AMQP、STOMP、MQTT 等等。
    • 支持多种语言客户端。RabbitMQ几乎支持所有常用编程语言，包括 Java、Python 、JavaScript、Ruby、Go等。
    • 插件机制。RabbitMQ提供了许多插件，可以通过插件进行扩展，也可以编写自己的插件。

    2. RabbitMQ的安装

      2.1 RabbitMQ下载 

          官网下载地址：https://www.rabbitmq.com/download.html

          我们可以根据自身需求和操作系统版本下载自己想要的版本，我现在选择的是Linux操作系统CentOS版本下的 rabbitmq-server-3.8.22-1.el7.noarch.rpm，同时也需要下载erlang语言的依赖 erlang-23.2.5-1.el7.x86_64.rpm

      2.2 下载的安装包

      2.3  安装步骤

            1. 上传rabbitmq的软件包到Linux目录       

     

            2.  先安装erlang的依赖包

       rmp -ivh erlang-23.2.5-1.el7.x86_64.rpm

            3.  再安装socat的依赖包

       rmp -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm

            4.  最后安装rabbitmq的依赖包

       rmp -ivh rabbitmq-server-3.8.22-1.el7.noarch.rpm

            5.  安装成功后再新建主配置文件rabbitmq.conf

       vim /etc/rabbitmq/rabbitmq.conf

            6.  rabbitmq.conf 配置文中添加如下配置

    1.    # 默认允许登录的用户名，无法修改
    2.    default_user = guest
    3.  
    4.    # 默认用户的密码，无法修改
    5.    default_pass = guest
    6.    
    7.    # 配置默认用户的身份为管理员
    8.    default_user_tags.administrator = true
    9.  
    10.    # 创建默认用户时分配给它的​权限​​
    11.    default_permissions.configure = .*
    12.    default_permissions.read = .*
    13.    default_permissions.write = .*
    14.    
    15.    # 默认loopback_users.guest = true，只允许本机访问，配置“guest”用户能够远程连接
    16.    # 注意，生产环境下不建议开启guest用户远程登录
    17.    loopback_users = none

             7. 用户管理

    1.    rabbitmqctl add_user admin 123                    # 创建新用户和密码  
    2.    
    3.    rabbitmqctl change_password admin admin123        # 修改用户密码
    4.    
    5.    rabbitmqctl set_user_tags admin administrator     # 设置角色   
    6.    
    7.    rabbitmqctl set_permissions -p / admin '.*' '.*' '.*'  # 设置用户权限

             8. 开启rabbitmq的管理界面插件功能

       rabbitmq-plugins enable rabbitmq_management

             9.  启动rabbitmq的服务

       systemctl start rabbitmq-server

    2.4  登录访问

         在浏览器中访问：192.168.8.129:15672 打开登录界面，输入用户名和密码：guest / guest 进行登录，登录成功后就可以进入到RabbitMQ的管理界面。

     

     3. RabbitMQ的入门

       3.1 RabbitMQ的架构原理

     RabbitMQ中的相关概念：

    •  Broker：RabbitMQ的实体服务器，提供一种传输服务，维护一条从生产者到消费者的传输线路，保证消息数据能按照指定的方式传输。
    • Virtual host：虚拟主机，与用户绑定，用于环境隔离。用户只能在自己的虚拟主机中创建Exchange和Queue，并且同一个虚拟主机里面不能有相同名称的Exchange或Queue。
    • Connection：Producer或Consumer和Broker之间建立的TCP连接。
    • Channel：Channel是在Connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的Channel进行通信，Channel作为轻量级的Connection极大减少了操作系统建立TCP Connection的开销。
    • Exchange：交换机，接收消息，按照路由规则将消息路由到一个或者多个队列。如果路由不到，或者返回给生产者，或者直接丢弃。RabbitMQ常用的交换器常用类型有direct、topic、fanout、headers四种。
    • Queue：接收交换机路由的消息的容器，消费者监听队列中有消息后进行消费。
    • Binding：绑定，Exchange和Queue之间的虚拟连接，绑定中可以包含一个或者多个routingKey。Binding信息被保存到Exchange中的查询表中，用于消息分发的依据。

      3.2 永远的hello world程序 

    •  添加Maven依赖
    1.  <dependency>
    2.      <groupId>com.rabbitmqgroupId>
    3.      <artifactId>amqp-clientartifactId>
    4.      <version>5.8.0version>
    5.  dependency>
    • 在rabbitmq的管理界面中新建虚拟主机 /study 

    •  书写RabbitMQUtils工具类
    1. package com.example.rabbitmq.common;
    2. import com.rabbitmq.client.Channel;
    3. import com.rabbitmq.client.Connection;
    4. import com.rabbitmq.client.ConnectionFactory;
    5. /**
    6.  * @author: gjm
    7.  * @date: 2021/9/2 13:50
    8.  * @description: TODO
    9.  */
    10. public class RabbitMQUtils {
    11.  
    12.     private static final ConnectionFactory connectionFactory;
    13.     static {
    14.         //创建连接mq的连接工厂对象
    15.         connectionFactory = new ConnectionFactory();
    16.         //设置连接rabbitmq主机
    17.         connectionFactory.setHost("192.168.8.129");
    18.         //设置端口号
    19.         connectionFactory.setPort(5672);
    20.         //设置连接哪个虚拟主机
    21.         connectionFactory.setVirtualHost("/study");
    22.         //设置访问虚拟主机的用户名和密码
    23.         connectionFactory.setUsername("guest");
    24.         connectionFactory.setPassword("guest");
    25.     }
    26.     /**
    27.      * 获取连接的方法
    28.      * @return
    29.      */
    30.     public static Connection getRabbitMQConnection(){
    31.         try{
    32.             return connectionFactory.newConnection();
    33.         }catch (Exception e){
    34.             e.printStackTrace();
    35.         }
    36.         return null;
    37.     }
    38.     /**
    39.      * 关闭通道和连接的方法
    40.      * @param channel
    41.      * @param conn
    42.      */
    43.     public static void closeConnectionAndChanel(Channel channel, Connection conn) {
    44.         try{
    45.             if(channel != null) channel.close();
    46.             if(conn != null) conn.close();
    47.         } catch (Exception e) {
    48.             e.printStackTrace();
    49.         }
    50.     }
    51. }
    •  书写生产者Provider 
    1. import com.example.rabbitmq.common.RabbitMQUtils;
    2. import com.rabbitmq.client.Channel;
    3. import com.rabbitmq.client.Connection;
    4. import java.io.IOException;
    5. /**
    6.  * @author: gjm
    7.  * @date: 2021/9/2 11:05
    8.  * @description: 生产者发送消息
    9.  */
    10. public class Provider {
    11.  
    12.     private static final String QUEUE_NAME = "hello";
    13.  
    14.     public static void main(String[] args) throws IOException{
    15.         //获取连接对象
    16.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    17.         //获取连接中通道
    18.         Channel channel = connection.createChannel();
    19.         //通道绑定对应消息队列
    20.         //参数1: 队列名称，如果队列不存在则自动创建
    21.         //参数2: 用来定义队列是否要持久化 true：持久化队列，false：不持久化
    22.         //参数3: 是否独占队列 true: 独占队列， false: 不独占
    23.         //参数4: autoDelete 是否在消费完成后自动删除队列， true:自动删除，false: 不自动删除
    24.         channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    25.         //发布消息
    26.         //参数1:交换机名称  参数2:队列名称  参数3:传递消息额外设置  参数4:消息的具体内容
    27.         channel.basicPublish("",QUEUE_NAME, null, "hello world".getBytes());
    28.         //关闭连接
    29.         RabbitMQUtils.closeConnectionAndChanel(channel, connection);
    30.     }
    31. }
    •  书写消费者Consumer
    1. import com.rabbitmq.client.*;
    2. import com.example.rabbitmq.common.RabbitMQUtils;
    3. import java.io.IOException;
    4. /**
    5.  * @author: guojm
    6.  * @date: 2021/9/3 0:47
    7.  * @description: 消费者消费消息
    8.  */
    9. public class Consumer {
    10.  
    11.     private static final String QUEUE_NAME = "hello";
    12.  
    13.     public static void main(String[] args) throws IOException{
    14.         //获取连接对象
    15.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    16.         //获取连接中通道
    17.         Channel channel = connection.createChannel();
    18.         //通道绑定对应消息队列
    19.         //参数1: queue: 队列名称，如果队列不存在则自动创建
    20.         //参数2: durable: 队列是否要持久化 true：持久化队列，false：不持久化
    21.         //参数3: exclusive: 是否独占队列 true: 独占队列， false: 不独占
    22.         //参数4: autoDelete 是否在消费完成后自动删除队列， true:自动删除，false: 不自动删除
    23.         channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    24.         System.out.println("等待消息中...");
    25.         // 消息消费时回调函数
    26.         DeliverCallback deliverCallback = (consumerTag, message)->{
    27.             System.out.println("消费消息：" + new String(message.getBody()));
    28.             System.out.println("等待消息中...");
    29.         };
    30.         // 消息取消时回调函数
    31.         CancelCallback cancelCallback = consumerTag ->{};
    32.         //发布消息
    33.         //参数1: 消费哪个队列的消息 队列名称
    34.         //参数2: 开始消息的自动确认机制
    35.         //参数3: 消息的接收回调
    36.         //参数3: 消息的取消回调
    37.         channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
    38.     }
    39. }
    •    先执行消费者没有消息时会等待，再执行生产者发送消息，消费者有消息时就会消费消息 

    4. RabbitMQ的工作模式

      4.1 简单模式(Hello World)

          在上图的简单模式中一个生产者(Producer)生产的消息只有一个消费者(Consumer)消费，Producer直接发送消息到hello队列，Consumer直接从hello队列中取出消息消费。  

    • 简单模式生产者代码为入门案例代码
    1.    //获取连接对象
    2.    Connection connection = RabbitMQUtils.getRabbitMQConnection();
    3.    //获取连接中通道
    4.    Channel channel = connection.createChannel();
    5.    //通道绑定对应消息队列
    6.    channel.queueDeclare("hello", true, false, false, null);
    7.    //生产者直接发送消息给队列
    8.    channel.basicPublish("","hello", null, "hello world".getBytes());
    9.    //关闭连接
    10.    RabbitMQUtils.closeConnectionAndChanel(channel, connection);
    • 消费者代码也是入门案例代码
    1.    //获取连接对象
    2.    Connection connection = RabbitMQUtils.getRabbitMQConnection();
    3.    //获取连接中通道
    4.    Channel channel = connection.createChannel();
    5.    //通道绑定对应消息队列
    6.    channel.queueDeclare("hello", true, false, false, null);
    7.    System.out.println("等待消息中...");
    8.    // 消息消费时回调函数
    9.    DeliverCallback deliverCallback = (consumerTag, message)->{
    10.        System.out.println("消费消息：" + new String(message.getBody()));
    11.        System.out.println("等待消息中...");
    12.    };
    13.    // 消息取消时回调函数
    14.    CancelCallback cancelCallback = consumerTag ->{};
    15.    // 消费者直接从队列中取出消息消费
    16.    channel.basicConsume("hello", true, deliverCallback, cancelCallback);

    4.2 工作队列模式(Work Queues) 

       在工作队列模式中一个生产者(Producer)生产的消息由多个消费者(Consumer)消费，这种情况适合生产者生产消息速度远远大于消息消费速度的场景。这种模式下可以让多个消费者绑定同一个队列，共同消费队列中的消息，多个消费者之间是竞争关系，消息不会被重复消费。

    •  工作队列的生产者代码
    1. import com.rabbitmq.client.Channel;
    2. import com.rabbitmq.client.Connection;
    3. import com.example.rabbitmq.common.RabbitMQUtils;
    4. import java.io.IOException;
    5. /**
    6.  * @author: gjm
    7.  * @date: 2021/9/04 22:03
    8.  * @description: Work Queues 模式的生产者
    9.  */
    10. public class Producer {
    11.  
    12.     private static final String QUEUE_NAME = "work_queue";
    13.  
    14.     public static void main(String[] args) throws IOException {
    15.         // 获取rabbitmq 的连接
    16.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    17.         // 通过连接获取信道
    18.         Channel channel = connection.createChannel();
    19.         // 声明队列
    20.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    21.         // 发布消息给队列
    22.         for (int i = 0; i < 10; i++) {
    23.             channel.basicPublish("",QUEUE_NAME, null, ("work queues 生产者的消息" + i ).getBytes());
    24.         }
    25.         // 关闭连接
    26.         RabbitMQUtils.closeConnectionAndChanel(channel,connection);
    27.     }
    28. }
    • 工作队列的消费者代码，创建消费者1 和消费者2共同消费。
    1. import com.rabbitmq.client.*;
    2. import com.example.rabbitmq.common.RabbitMQUtils;
    3. import java.io.IOException;
    4. /**
    5.  * @author: gjm
    6.  * @date: 2021/9/04 22:11
    7.  * @description: Work Queues 模式的消费者
    8.  */
    9. public class Consumer1 {
    10.     private static final String QUEUE_NAME = "work_queue";
    11.  
    12.     public static void main(String[] args) throws IOException {
    13.         // 获取rabbitmq的连接对象
    14.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    15.         // 通过连接对象获取信道
    16.         Channel channel = connection.createChannel();
    17.         // 声明队列，如果队列不存在则自动创建
    18.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    19.         // 消息消费时回调函数
    20.         DeliverCallback deliverCallback = (consumerTag, message)-> {
    21.             System.out.println("消费者1：" + new String(message.getBody()));
    22.         };
    23.         // 获取队列中的消息
    24.         channel.basicConsume(QUEUE_NAME, true, deliverCallback,consumerTag ->{});
    25.     }
    26. }
    • 先运行消费者1和消费者2代码，再运行生产者代码查看消费情况。从下面的结果我们可以看出工作队列模式中的消息默认是循环分发给每一个消费者的，就是不管消费者的消费速度如何都会按照轮询分发给每一个消费者。

     

    • 工作队列模式中的代码优化，消费者端不开启消息的自动确认机制，autoAck设置为fasle，开启消息的自动确认机制可能会导致消息丢失。因为消费者端开启消息的自动确认机制后消息会轮询分发给消费者等待消费，不管消息是否消费成功都会立马在队列中删除，这时如果消费端宕机就会导致消息丢失。
    1. import com.rabbitmq.client.*;
    2. import com.example.rabbitmq.common.RabbitMQUtils;
    3. import java.io.IOException;
    4. /**
    5.  * @author: gjm
    6.  * @date: 2021/9/04 22:11
    7.  * @description: Work Queues 消费者手动确认消息
    8.  */
    9. public class Consumer1 {
    10.     private static final String QUEUE_NAME = "work_queue";
    11.  
    12.     public static void main(String[] args) throws IOException {
    13.         // 获取rabbitmq的连接对象
    14.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    15.         // 通过连接对象获取信道
    16.         Channel channel = connection.createChannel();
    17.         // 声明队列，如果队列不存在则自动创建
    18.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    19.         // 消息消费时回调函数
    20.         DeliverCallback deliverCallback = (consumerTag, message)-> {
    21.             System.out.println("消费者1：" + new String(message.getBody()));
    22.             // 手动确认消息
    23.             // 参数1：被确认消息的标签信息，参数2：是否批量确认消息
    24.             channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
    25.         };
    26.         // 消费队列中的消息，关闭消息的自动确认机制，autoAck设置为false
    27.         channel.basicConsume(QUEUE_NAME, false, deliverCallback,consumerTag ->{});
    28.     }
    29. }
    • 工作队列模式实现消费端能者多劳，即消费者消费快的能多消费消息，实现消息的不公平分发。这时只需要在消费者端设置信道每次只能接受一个消息等待消费，消费者消费快的就能多接收消息消费。
    1.         // 消费者端设置信道每次只能接收一个消息
    2.         channel.basicQos(1);
    3.         // 消息消费时回调函数
    4.         DeliverCallback deliverCallback = (consumerTag, message)-> {
    5.             //模拟消息消费慢的场景
    6.             try{
    7.                 Thread.sleep(500);
    8.             }catch (Exception e){
    9.                 e.printStackTrace();
    10.             }
    11.             System.out.println("消费者1：" + new String(message.getBody()));
    12.             // 手动确认消息
    13.             // 参数1：被确认消息的标签信息，参数2：是否批量确认消息
    14.             channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
    15.         };
    16.         // 消费队列中的消息，关闭消息的自动确认机制，autoAck设置为false
    17.         channel.basicConsume(QUEUE_NAME, false, deliverCallback,consumerTag ->{});
    •  工作队列模式中的能者多劳效果实现

     

     4.3 发布订阅模式(Publish/Subscribe) 

    ​

         发布订阅模式也被称为广播模式，就是使用广播（fanout）类型的交换机实现的。生产者直接发送消息给交换机，由交换机负责将消息转发给绑定的多个队列实现被多个消费者消费的效果。这个模式实现一个消息被多个消费者消费的原理是用到了多个临时队列，fanout型交换机把一个消息转发到了多个队列中，每个队列的消费者各自消费自己绑定队列中的消息。

    • 发布订阅模式的生产者代码 
    1. import com.example.rabbitmq.common.RabbitMQUtils;
    2. import com.rabbitmq.client.Channel;
    3. import com.rabbitmq.client.Connection;
    4. import java.io.IOException;
    5. /**
    6.  * @author: gjm
    7.  * @date: 2021/9/2 21:06
    8.  * @description: Publish/Subscribe模型的生产者
    9.  */
    10. public class Producer {
    11.  
    12.     public static void main(String[] args)throws IOException {
    13.         //获取连接对象
    14.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    15.         Channel channel = connection.createChannel();
    16.         //通过通道声明指定交换机
    17.         // 参数1：交换机名称   参数2：交换机类型  fanout广播类型
    18.         channel.exchangeDeclare("logs", "fanout");
    19.         //发布消息给交换机
    20.         channel.basicPublish("logs", "",null, "Publish/Subscribe 生产者的消息".getBytes());
    21.         //关闭连接
    22.         RabbitMQUtils.closeConnectionAndChanel(channel, connection);
    23.     }
    24. }
    • 发布订阅模式的消费者代码，创建消费者1和消费者2演示消息的广播效果
    1. import com.example.rabbitmq.common.RabbitMQUtils;
    2. import com.rabbitmq.client.*;
    3. import java.io.IOException;
    4. /**
    5.  * @author: gjm
    6.  * @date: 2021/9/2 21:30
    7.  * @description: Publish/Subscribe模型的消费者1
    8.  */
    9. public class consumer1 {
    10.  
    11.     public static void main(String[] args)throws IOException {
    12.         //获取连接对象
    13.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    14.         //获取通道
    15.         Channel channel = connection.createChannel();
    16.         //通道绑定交换机
    17.         channel.exchangeDeclare("logs", "fanout");
    18.         //临时队列
    19.         String queueName = channel.queueDeclare().getQueue();
    20.         //绑定交换机和临时队列
    21.         channel.queueBind(queueName, "logs", "");
    22.         //消费消息
    23.         channel.basicConsume(queueName, true, new DefaultConsumer(channel){
    24.             @Override
    25.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    26.                 System.out.println("消费者1：" + new String(body));
    27.             }
    28.         });
    29.     }
    30. }
    • 先运行消费者在管理界面中查看创建的临时队列信息

    • 消费者1和消费者2都消费到了生产者发送的消息

     

     4.4 路由模式(Routing) 

     

          路由模式是由direct类型的交换机实现的，交换机和队列通过routingKey进行绑定，生产者发送消息给交换机时需要指定routingKey，交换机在转发消息给队列时会根据routingKey直接找到队列转发。路由模式下同一个routingKey可以和多个队列绑定，所以也能实现广播模式的效果，但是业务中一般是消息是需要由不同的标识做区别处理场景下使用。

    • 路由模式的生产者代码 
    1. import com.example.rabbitmq.common.RabbitMQUtils;
    2. import com.rabbitmq.client.Channel;
    3. import com.rabbitmq.client.Connection;
    4. import java.io.IOException;
    5. /**
    6.  * @author: gjm
    7.  * @date: 2021/9/3 9:50
    8.  * @description: 路由模式的生产者
    9.  */
    10. public class Producer {
    11.     
    12.     public static final String ROUTING_KEY = "error";
    13.     
    14.     public static void main(String[] args) throws IOException {
    15.         //获取连接对象
    16.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    17.         //获取连接中的信道
    18.         Channel channel = connection.createChannel();
    19.         //通过信道声明交换机名称和类型
    20.         channel.exchangeDeclare("logs direct", "direct");
    21.         //发送消息给交换机并指定routingKey
    22.         channel.basicPublish("logs direct", ROUTING_KEY,null, ("路由模型生产者发布的routingKey:"+ ROUTING_KEY +"的消息").getBytes());
    23.         //关闭连接
    24.         RabbitMQUtils.closeConnectionAndChanel(channel,connection);
    25.     }
    26. }
    •  路由模式的消费者代码 
    1. import com.example.rabbitmq.common.RabbitMQUtils;
    2. import com.rabbitmq.client.*;
    3. import java.io.IOException;
    4. /**
    5.  * @author: gjm
    6.  * @date: 2021/9/3 10:04
    7.  * @description: 路由模式的消费者
    8.  */
    9. public class Consumer1 {
    10.     public static void main(String[] args) throws IOException {
    11.         //获取连接对象
    12.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    13.         //获取连接的信道对象
    14.         Channel channel = connection.createChannel();
    15.         //通过信道声明交换机名称和类型
    16.         channel.exchangeDeclare("logs direct", "direct");
    17.         //创建一个临时队列
    18.         String queue = channel.queueDeclare().getQueue();
    19.         //基于routingkey绑定队列和交换机
    20.         channel.queueBind(queue, "logs direct", "error");
    21.         //获取消费的消息
    22.         channel.basicConsume(queue, true, new DefaultConsumer(channel){
    23.             @Override
    24.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    25.                 System.out.println("消费者1："+ new String(body));
    26.             }
    27.         });
    28.     }
    29. }
    •  消费者1程序运行，生产者发送routingKey为error的消息只能被routingKey也为error的消费者消费

    4.5 主题模式(Topics) 

      

         主题模式本质就是路由模式的升级，也被称为动态路由模式。消费者端在声明队列和topic类型的交换机绑定时采用通配符形式的routingKey达到动态匹配的效果，这种模型routingKey一般是由一个或多个单词组成，多个单词之间以 “.” 分割，例如 order.create

    • 通配符 * 只能代替一个词
      比如：user.* 只能匹配 user.create 或 user.update这种.后面一个单词的
    •  通配符 # 可以替换零个或多个单词
      比如：order.# 能匹配 order.seckill.create 或 order.delete
    •  主题模式的生产者代码 
    1. import com.example.rabbitmq.common.RabbitMQUtils;
    2. import com.rabbitmq.client.Channel;
    3. import com.rabbitmq.client.Connection;
    4. import java.io.IOException;
    5. /**
    6.  * @author: gjm
    7.  * @date: 2021/9/3 10:38
    8.  * @description: topics模式的生产者
    9.  */
    10. public class Producer {
    11.  
    12.     public static final String ROUTING_KEY = "order.seckill.create";
    13.  
    14.     public static void main(String[] args) throws IOException {
    15.         //获取连接对象
    16.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    17.         //获取连接的信道对象
    18.         Channel channel = connection.createChannel();
    19.         //通过信道声明交换机名称和类型
    20.         channel.exchangeDeclare("topics", "topic");
    21.         //发布消息
    22.         channel.basicPublish("topics", ROUTING_KEY,null, ("主题模式生产者发布的routingKey:"+ ROUTING_KEY +"的消息").getBytes());
    23.         //关闭连接
    24.         RabbitMQUtils.closeConnectionAndChanel(channel,connection);
    25.     }
    26. }
    • 主题模式的消费者代码 
    1. import com.example.rabbitmq.common.RabbitMQUtils;
    2. import com.rabbitmq.client.*;
    3. import java.io.IOException;
    4. /**
    5.  * @author: gjm
    6.  * @date: 2021/9/3 10:51
    7.  * @description: topics模式的消费者
    8.  */
    9. public class Consumer1 {
    10.  
    11.     public static void main(String[] args) throws IOException{
    12.         //获取连接对象
    13.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    14.         //获取连接的信道对象
    15.         Channel channel = connection.createChannel();
    16.         //通过信道声明交换机名称和类型
    17.         channel.exchangeDeclare("topics", "topic");
    18.         //通过信道声明队列
    19.         channel.queueDeclare("Q1", true, false, false, null);
    20.         //基于通配符的routingkey绑定队列和交换机
    21.         channel.queueBind("Q1", "topics", "order.#");
    22.         //消费消息
    23.         channel.basicConsume("Q1", true, new DefaultConsumer(channel){
    24.             @Override
    25.             public void handleDelivery(String consumerTag, Envelope envelope, 
    26.                 AMQP.BasicProperties properties, byte[] body) throws IOException {
    27.                 System.out.println("消费者1："+ new String(body));
    28.             }
    29.         });
    30.     }
    31. }
    • 消费者1程序运行，生产者发送routingKey为order.seckill.create的消息被routingKey为order.#的消费者消费

    4.6 发布确认模式(Publisher Confirms)  

         

          发布确认模式和前面介绍的模式有所不同，它更像是一种生产者端保证消息能够进行可靠投递的一种发送策略。如何保证消息不丢失这是生产中经常需要解决的问题，我们一般需要对队列和消息进行开启持久化机制，但是还需要考虑生产者端和消费者端的消息丢失问题，而发布确认模式是用于解决生产者发布消息到MQ服务器的过程中的消息丢失问题。从上图可以看出 RabbitMQ 提供的是“至少一次交付”（at-least-once delivery），异常情况下，消息会被重复投递或消费。

    • 单个消息发布确认策略，即每次发布一个消息后都等待MQ服务端应答确认成功后再继续发下一个消息；这种方式消息投递的效率不高，但是比较可靠。
    1. import com.rabbitmq.client.Channel;
    2. import com.rabbitmq.client.Connection;
    3. import com.rabbitmq.client.MessageProperties;
    4. import com.example.rabbitmq.common.RabbitMQUtils;
    5.  
    6. import java.io.IOException;
    7. import java.util.UUID;
    8. /**
    9.  * @author: gjm
    10.  * @date: 2021/9/05 8:50
    11.  * @description: 单个消息发布确认策略
    12.  */
    13. public class PublishMessageIndividually {
    14.  
    15.     public static void main(String[] args) throws IOException, InterruptedException {
    16.         // 获取rabbitmq的连接
    17.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    18.         // 通过连接创建信道
    19.         Channel channel = connection.createChannel();
    20.         // 生成队列名称
    21.         String queueName = UUID.randomUUID().toString();
    22.         // 声明队列,开启队列持久化
    23.         channel.queueDeclare(queueName, true, false,false, null);
    24.         // 开启发布确认
    25.         channel.confirmSelect();
    26.         // 开始时间
    27.         long begin = System.currentTimeMillis();
    28.         int count = 0;
    29.         // 批量发消息
    30.         for (int i = 0; i < 1000; i++) {
    31.             String message ="单个确认模式的消息: "+i;
    32.             // 投递消息，设置消息的持久化
    33.             channel.basicPublish("",queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    34.             // 单个消息发送后，调用同步等待确认方法获取结果
    35.             boolean flag = channel.waitForConfirms();
    36.             // 如果flag为true则表明消息被确认成功投递
    37.             if(flag) count++;
    38.         }
    39.         // 如果确认的消息数等于发布数则消息被全部投递成功
    40.         if (count == 1000) {
    41.             System.out.println("成功投递"+ count + "个消息并且确认");
    42.         }
    43.         //关闭连接
    44.         RabbitMQUtils.closeConnectionAndChanel(channel, connection);
    45.         long end = System.currentTimeMillis();
    46.         System.out.println("投递"+ 1000 + "个单个确认的消息，耗时："+(end-begin)+"ms");
    47.  
    48.     }
    49. }

       单个消息发布确认策略运行结果： 

    •  批量消息发布确认策略，即每次发布一批消息后才去请求MQ服务端的确认应答结果。这种方式要保证这一批消息全部都成功发送才能拿到确认成功的结果，中间只要有一条发送不成功拿到的是全局不成功的结果false。官网上已经说明这样会导致我们不知道哪条信息没有被成功发送，需要记录这批消息的内容做消息的发送重试，这会导致消息的重复发送，消费者端要处理消息的幂等性问题。
    1. import com.rabbitmq.client.Channel;
    2. import com.rabbitmq.client.Connection;
    3. import com.rabbitmq.client.MessageProperties;
    4. import com.example.rabbitmq.common.RabbitMQUtils;
    5.  
    6. import java.io.IOException;
    7. import java.util.UUID;
    8. /**
    9.  * @author: gjm
    10.  * @date: 2021/9/05 9:57
    11.  * @description: 消息批量确认策略
    12.  */
    13. public class PublishMessageBatch {
    14.  
    15.     public static void main(String[] args) throws IOException, InterruptedException {
    16.         // 获取rabbitmq的连接
    17.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    18.         // 通过连接创建信道
    19.         Channel channel = connection.createChannel();
    20.         // 生成队列名称
    21.         String queueName = UUID.randomUUID().toString();
    22.         // 声明队列,开启队列持久化
    23.         channel.queueDeclare(queueName, true, false,false, null);
    24.         // 开启发布确认
    25.         channel.confirmSelect();
    26.         // 开始时间
    27.         long begin = System.currentTimeMillis();
    28.         // 每次确认需要投递的消息总数
    29.         int batchSize = 100;
    30.         // 记录已投递的消息总数
    31.         int outstandingMessageCount = 0;
    32.         // 记录确认的次数
    33.         int confirmCount = 0;
    34.         // 批量发消息
    35.         for (int i = 0; i < 1000; i++) {
    36.             String message ="批量确认模式的消息: "+i;
    37.             // 投递消息，设置消息的持久化
    38.             channel.basicPublish("",queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    39.             // 已投递的消息+1
    40.             outstandingMessageCount ++;
    41.             // 已投递的消息总数达到确认要求
    42.             if (outstandingMessageCount == batchSize) {
    43.                 outstandingMessageCount = 0;
    44.                 confirmCount ++;
    45.                 boolean confirms = channel.waitForConfirms();
    46.                 if (confirms) {
    47.                     System.out.println("第" +confirmCount+ "次投递"+ batchSize +"条消息并确认");
    48.                 }
    49.             }
    50.         }
    51.         RabbitMQUtils.closeConnectionAndChanel(channel, connection);
    52.         long end = System.currentTimeMillis();
    53.         System.out.println("投递"+ 1000 + "个批量确认消息，耗时："+(end-begin)+"ms");
    54.     }
    55. }

     批量消息发布确认策略运行结果： 

    • 异步消息发布确认策略，即消息的发布和确认是异步的，可以显著提升消息发布的速度，也能通过异步消息确认的回调函数记录出未被确认投递成功的消息，三种消息确认发布策略中官网也指出这种效率为最高。
    1. import com.rabbitmq.client.Channel;
    2. import com.rabbitmq.client.ConfirmCallback;
    3. import com.rabbitmq.client.Connection;
    4. import com.example.rabbitmq.common.RabbitMQUtils;
    5.  
    6. import java.io.IOException;
    7. import java.util.concurrent.ConcurrentNavigableMap;
    8. import java.util.concurrent.ConcurrentSkipListMap;
    9. /**
    10.  * @author: gjm
    11.  * @date: 2021/9/05 11:12
    12.  * @description: 异步消息发布确认策略 
    13.  */
    14. public class PublishMessageAsync {
    15.  
    16.     public static void main(String[] args) throws IOException{
    17.         // 获取rabbitmq的连接
    18.         Connection connection = RabbitMQUtils.getRabbitMQConnection();
    19.         // 通过连接创建信道
    20.         Channel channel = connection.createChannel();
    21.         // 声明队列,开启队列持久化
    22.         String queueName = "confirm_queue";
    23.         channel.queueDeclare(queueName, true, false,false, null);
    24.         // 开启发布确认
    25.         channel.confirmSelect();
    26.         // 开始时间
    27.         long begin = System.currentTimeMillis();
    28.  
    29.         // 记录未确认的消息，方便后续重发
    30.         ConcurrentSkipListMap outstandingConfirms = new ConcurrentSkipListMap<>();
    31.         //消息确认成功的回调函数
    32.         ConfirmCallback ackCallback = ((deliveryTag, multiple) -> {
    33.             /*
    34.              * 如果multiple等于true表明这次回调为多个消息确认，标识为deliveryTag的消息和他前面的消息都一起被确认
    35.              * 如果multiple等于false表明这次回调为单个消息确认，只有标识为deliveryTag的消息被确认
    36.              */
    37.             if (multiple) {
    38.                 // 使用headMap方法拿到标识为deliveryTag的消息和他前面的消息
    39.                 ConcurrentNavigableMap confirmed = outstandingConfirms.headMap(deliveryTag, true);
    40.                 // 批量删除已确认的消息
    41.                 confirmed.clear();
    42.             } else {
    43.                 // 删除单个确认的消息
    44.                 outstandingConfirms.remove(deliveryTag);
    45.             }
    46.             System.out.println("多个消息确认："+ multiple + " -- 确认的消息deliveryTag："+ deliveryTag);
    47.         });
    48.         //消息未确认的回调函数
    49.         ConfirmCallback nackCallback = ((deliveryTag, multiple) -> {
    50.             // 取出未确认的消息
    51.             String message = outstandingConfirms.get(deliveryTag);
    52.             // 打印消息的详情
    53.             System.out.println("message："+ message+ " -- deliveryTag：" +deliveryTag+ " -- multiple："+multiple);
    54.         });
    55.         // 消息的监听器，监听哪些消息成功和失败
    56.         channel.addConfirmListener(ackCallback, nackCallback);
    57.         // 批量发消息
    58.         for (int i = 0; i < 1000; i++) {
    59.             String message ="批量异步确认模式的消息: "+i;
    60.             channel.basicPublish("",queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    61.             // 保存未处理的确认信息
    62.             outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
    63.         }
    64.  
    65.         long end = System.currentTimeMillis();
    66.         System.out.println("投递"+ 1000 + "个异步批量确认消息，耗时："+(end-begin)+"ms");
    67.     }
    68. }

     异步消息发布确认策略运行结果：

     

    5. RabbitMQ的注解开发

       5.1 简单模式

    •  添加springboot整合rabbitmq的依赖
    1.  <dependency>
    2.      <groupId>org.springframework.boot</groupId>
    3.      <artifactId>spring-boot-starter-amqp</artifactId>
    4.  </dependency>
    •  在application.yml文件中配置如下内容
    1.  spring:
    2.    rabbitmq:
    3.       host: 192.168.8.129
    4.       port: 5672
    5.       username: guest
    6.       password: guest
    7.       virtual-host: /study
    •  书写生产者的代码，使用接口开发，在SpringBoot程序的主入口类上书写注解@EnableRabbit ，表示开启rabbitmq的自动配置功能，然后就可以直接注入和使用RabbitTemplate对象非常简单地发送消息给指定队列。
    1. import org.slf4j.Logger;
    2. import org.slf4j.LoggerFactory;
    3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    4. import org.springframework.beans.factory.annotation.Autowired;
    5. import org.springframework.web.bind.annotation.RequestMapping;
    6. import org.springframework.web.bind.annotation.RestController;
    7. /**
    8.  * @author: gjm
    9.  * @date: 2021/9/04 10:14
    10.  * @description: 生产者端
    11.  */
    12. @RestController
    13. @RequestMapping("/producer")
    14. public class ProducerController {
    15.     private Logger logger = LoggerFactory.getLogger(ProducerController.class);
    16.     @Autowired
    17.     private RabbitTemplate rabbitTemplate;
    18.  
    19.     @RequestMapping("/simple")
    20.     public String sendSimpleMessage(){
    21.         String message = "Hello World";
    22.         rabbitTemplate.convertAndSend("hello", message);
    23.         logger.info("简单模式的生产者发送了消息：{}", message);
    24.         return "success";
    25.     }
    26. }
    • 书写消费者的代码，消费者端使用@RabbitListener注解来声明一个队列，表示这是消费者端。通过@Queue(value = "hello")指定的队列名称，并且可以指定队列持久化，独占，自动删除等参数，默认不指定的话队列是持久化，非独占，非自动删除的。
    1. import org.slf4j.Logger;
    2. import org.slf4j.LoggerFactory;
    3. import org.springframework.amqp.rabbit.annotation.Queue;
    4. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    6. import org.springframework.stereotype.Component;
    7. /**
    8.  * @author: gjm
    9.  * @date: 2021/9/04 9:40
    10.  * @description: 消费者端
    11.  */
    12. @Component
    13. @RabbitListener(queuesToDeclare = @Queue(value = "hello"))
    14. public class Consumer {
    15.     private Logger logger = LoggerFactory.getLogger(Consumer.class);
    16.  
    17.     @RabbitHandler
    18.     public void receive(String message){
    19.         logger.info("简单模式的消费者接收了消息：{}", message);
    20.     }
    21. }
    • 运行程序，调用接口/producer/simple发送消息，消费者端会监听到消息并处理。

     

     5.2 工作队列

    •  加入配置
    1.  spring:
    2.    rabbitmq:
    3.      listener:
    4.       simple:
    5.         prefetch: 1                 # 消费预取值，设置为1可实现消费者能者多劳，不设置则为轮询分发
    6.         acknowledge-mode: manual    # 消息需消费者手动确认
    • 生产者代码
    1. import org.slf4j.Logger;
    2. import org.slf4j.LoggerFactory;
    3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    4. import org.springframework.beans.factory.annotation.Autowired;
    5. import org.springframework.web.bind.annotation.RequestMapping;
    6. import org.springframework.web.bind.annotation.RestController;
    7. /**
    8.  * @author: gjm
    9.  * @date: 2021/9/04 10:14
    10.  * @description: 生产者端
    11.  */
    12. @RestController
    13. @RequestMapping("/producer")
    14. public class ProducerController {
    15.     private Logger logger = LoggerFactory.getLogger(ProducerController.class);
    16.     @Autowired
    17.     private RabbitTemplate rabbitTemplate;
    18.  
    19.     @RequestMapping("/work")
    20.     public String sendWorkMessage(){
    21.         for (int i = 0; i < 10; i++) {
    22.             String message = "work mode message " + i;
    23.             rabbitTemplate.convertAndSend("work", message);
    24.         }
    25.         logger.info("工作模式的生产者发送了10条消息");
    26.         return "success";
    27.     }
    28. }
    •  消费者代码
    1. import com.rabbitmq.client.Channel;
    2. import org.slf4j.Logger;
    3. import org.slf4j.LoggerFactory;
    4. import org.springframework.amqp.core.Message;
    5. import org.springframework.amqp.rabbit.annotation.Queue;
    6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    7. import org.springframework.stereotype.Component;
    8.  
    9. import java.io.IOException;
    10. import java.util.concurrent.TimeUnit;
    11. /**
    12.  * @author: gjm
    13.  * @date: 2021/9/04 9:15
    14.  * @description: 工作队列消费者
    15.  */
    16. @Component
    17. public class WorkConsumer {
    18.     private Logger logger = LoggerFactory.getLogger(WorkConsumer.class);
    19.  
    20.     @RabbitListener(queuesToDeclare = @Queue(value = "work"))
    21.     public void consumer1(String body, Message message, Channel channel) throws InterruptedException{
    22.         logger.info("工作模式的消费者1接收了消息：{}", body);
    23.         TimeUnit.MILLISECONDS.sleep(500);
    24.         try {
    25.             // 消费者手动确认消息已消费
    26.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    27.         } catch (IOException e) {
    28.             e.printStackTrace();
    29.         }
    30.     }
    31.     @RabbitListener(queuesToDeclare = @Queue(value = "work"))
    32.     public void consumer2(String body, Message message, Channel channel) {
    33.         logger.info("工作模式的消费者2接收了消息：{}", body);
    34.         try {
    35.             // 消费者手动确认消息已消费
    36.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    37.         } catch (IOException e) {
    38.             e.printStackTrace();
    39.         }
    40.     }
    41. }

    5.3 发布订阅

    •  使用配置类的方式声明队列和交换机并进行绑定(Binding)
    1. import org.springframework.amqp.core.*;
    2. import org.springframework.context.annotation.Bean;
    3. import org.springframework.context.annotation.Configuration;
    4. /**
    5.  * @author: gjm
    6.  * @date: 2021/9/05 8:16
    7.  * @description: 发布订阅模式的配置类
    8.  */
    9. @Configuration
    10. public class FanoutModeConfig {
    11.     // 交换机名称
    12.     public static final String FANOUT_EXCHANGE_NAME = "fanout_exchange";
    13.     // 队列名称
    14.     public static final String FANOUT_QUEUE1_NAME = "fanout_queue1";
    15.     // 队列名称
    16.     public static final String FANOUT_QUEUE2_NAME = "fanout_queue2";
    17.  
    18.     // 创建fanout类型的交换机
    19.     @Bean
    20.     public FanoutExchange fanoutExchange() {
    21.         return new FanoutExchange(FANOUT_EXCHANGE_NAME);
    22.     }
    23.     // 声明队列
    24.     @Bean
    25.     public Queue fanoutQueue1() {
    26.         return QueueBuilder.durable(FANOUT_QUEUE1_NAME).build();
    27.     }
    28.     // 声明队列
    29.     @Bean
    30.     public Queue fanoutQueue2() {
    31.         return QueueBuilder.durable(FANOUT_QUEUE2_NAME).build();
    32.     }
    33.     // 队列fanoutQueue1和交换机绑定
    34.     @Bean
    35.     public Binding queue1BindingExchange(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
    36.         return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    37.     }
    38.     // 队列fanoutQueue2和交换机绑定
    39.     @Bean
    40.     public Binding queue2BindingExchange(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
    41.         return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    42.     }
    43. }
    • 生产者代码，生产者直接发送消息给交换机。
    1. import org.slf4j.Logger;
    2. import org.slf4j.LoggerFactory;
    3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    4. import org.springframework.beans.factory.annotation.Autowired;
    5. import org.springframework.web.bind.annotation.RequestMapping;
    6. import org.springframework.web.bind.annotation.RestController;
    7. /**
    8.  * @author: gjm
    9.  * @date: 2021/9/05 10:14
    10.  * @description: 生产者端
    11.  */
    12. @RestController
    13. @RequestMapping("/producer")
    14. public class ProducerController {
    15.     private Logger logger = LoggerFactory.getLogger(ProducerController.class);
    16.     @Autowired
    17.     private RabbitTemplate rabbitTemplate;
    18.  
    19.     @RequestMapping("/fanout")
    20.     public String sendFanoutMessage(){
    21.         String message = "fanout mode message";
    22.         rabbitTemplate.convertAndSend("fanout_exchange", "", message);
    23.         logger.info("广播模式的生产者发送了消息：{}", message);
    24.         return "success";
    25.     }
    26. }
    •  消费者代码，消费者端使用@RabbitListener(queues = "queueName")直接监听队列就行，这里无需再声明和创建队列，配置类中已经声明。
    1. import com.rabbitmq.client.Channel;
    2. import org.slf4j.Logger;
    3. import org.slf4j.LoggerFactory;
    4. import org.springframework.amqp.core.Message;
    5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    6. import org.springframework.stereotype.Component;
    7. import java.io.IOException;
    8. /**
    9.  * @author: gjm
    10.  * @date: 2021/9/05 9:15
    11.  * @description: 发布订阅模式的消费者
    12.  */
    13. @Component
    14. public class FanoutConsumer {
    15.     private Logger logger = LoggerFactory.getLogger(FanoutConsumer.class);
    16.  
    17.     @RabbitListener(queues = "fanout_queue1")
    18.     public void consumer1(String body, Message message, Channel channel) {
    19.         logger.info("广播模式的消费者1接收了消息：{}", body);
    20.         try {
    21.             // 消费者手动确认消息已消费
    22.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    23.         } catch (IOException e) {
    24.             e.printStackTrace();
    25.         }
    26.     }
    27.     @RabbitListener(queues = "fanout_queue2")
    28.     public void consumer2(String body, Message message, Channel channel){
    29.         logger.info("广播模式的消费者2接收了消息：{}", body);
    30.         try {
    31.             // 消费者手动确认消息已消费
    32.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    33.         } catch (IOException e) {
    34.             e.printStackTrace();
    35.         }
    36.     }
    37. }

     5.4 路由模式

    • 使用direct类型的交换机通过routingKey和队列进行绑定(Binding) 
    1. import org.springframework.amqp.core.*;
    2. import org.springframework.context.annotation.Bean;
    3. import org.springframework.context.annotation.Configuration;
    4. /**
    5.  * @author: gjm
    6.  * @date: 2021/9/05 8:16
    7.  * @description: 路由模式的配置类
    8.  */
    9. @Configuration
    10. public class DirectModeConfig {
    11.     // 交换机名称
    12.     public static final String DIRECT_EXCHANGE_NAME = "direct_exchange";
    13.     // 队列名称
    14.     public static final String DIRECT_QUEUE_NAME = "direct_queue";
    15.  
    16.     // 创建direct类型的交换机
    17.     @Bean
    18.     public DirectExchange directExchange(){
    19.         return new DirectExchange(DIRECT_EXCHANGE_NAME);
    20.     }
    21.     // 声明队列
    22.     @Bean
    23.     public Queue directQueue(){
    24.         return QueueBuilder.durable(DIRECT_QUEUE_NAME).build();
    25.     }
    26.     // 队列directQueue和交换机使用routingKey绑定
    27.     @Bean
    28.     public Binding queueBindingExchange(Queue directQueue, DirectExchange directExchange){
    29.         return BindingBuilder.bind(directQueue).to(directExchange).with("route.test");
    30.     }
    31.  
    32. }
    • 生产者代码，生产者发送消息给交换机时指定routingKey。
    1. import org.slf4j.Logger;
    2. import org.slf4j.LoggerFactory;
    3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    4. import org.springframework.beans.factory.annotation.Autowired;
    5. import org.springframework.web.bind.annotation.GetMapping;
    6. import org.springframework.web.bind.annotation.RequestMapping;
    7. import org.springframework.web.bind.annotation.RestController;
    8. /**
    9.  * @author: gjm
    10.  * @date: 2021/9/05 19:14
    11.  * @description: TODO
    12.  */
    13. @RestController
    14. @RequestMapping("/producer")
    15. public class ProducerController {
    16.     private Logger logger = LoggerFactory.getLogger(ProducerController.class);
    17.  
    18.     @Autowired
    19.     private RabbitTemplate rabbitTemplate;
    20.  
    21.     @GetMapping("/routing")
    22.     public String sendRoutingMessage(){
    23.         String message = "routing mode message";
    24.         rabbitTemplate.convertAndSend("direct_exchange","route.test", message);
    25.         logger.info("路由模式的生产者发送了消息：{}", message);
    26.         return "success";
    27.     }
    28. }
    • 消费者代码
    1. import com.rabbitmq.client.Channel;
    2. import org.slf4j.Logger;
    3. import org.slf4j.LoggerFactory;
    4. import org.springframework.amqp.core.Message;
    5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    6. import org.springframework.stereotype.Component;
    7.  
    8. import java.io.IOException;
    9. /**
    10.  * @author: gjm
    11.  * @date: 2021/9/05 21:15
    12.  * @description: 路由模式的消费者
    13.  */
    14. @Component
    15. public class DirectConsumer {
    16.     private Logger logger = LoggerFactory.getLogger(DirectConsumer.class);
    17.  
    18.     @RabbitListener(queues = "direct_queue")
    19.     public void consumer(String body, Message message, Channel channel) {
    20.         logger.info("路由模式的消费者接收了消息：{}", body);
    21.         try {
    22.             // 消费者手动确认消息已消费
    23.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    24.         } catch (IOException e) {
    25.             e.printStackTrace();
    26.         }
    27.     }
    28. }

    5.5 主题模式

    • 使用topic类型的交换机通过routingKey和队列进行绑定(Binding) 
    1. import org.springframework.amqp.core.*;
    2. import org.springframework.context.annotation.Bean;
    3. import org.springframework.context.annotation.Configuration;
    4. /**
    5.  * @author: gjm
    6.  * @date: 2021/9/05 8:16
    7.  * @description: 主题模式的配置类
    8.  */
    9. @Configuration
    10. public class TopicModeConfig {
    11.     // 交换机名称
    12.     public static final String TOPIC_EXCHANGE_NAME = "topic_exchange";
    13.     // 队列名称
    14.     public static final String TOPIC_QUEUE_NAME = "topic_queue";
    15.  
    16.     // 创建topic类型的交换机
    17.     @Bean
    18.     public TopicExchange topicExchange(){
    19.         return new TopicExchange(TOPIC_EXCHANGE_NAME);
    20.     }
    21.     // 声明队列
    22.     @Bean
    23.     public Queue topicQueue(){
    24.         return QueueBuilder.durable(TOPIC_QUEUE_NAME).build();
    25.     }
    26.  
    27.     // 队列topicQueue和交换机使用routingKey绑定
    28.     @Bean
    29.     public Binding topicQueueBindingExchange(Queue topicQueue, TopicExchange topicExchange){
    30.         return BindingBuilder.bind(topicQueue).to(topicExchange).with("topic.*");
    31.     }
    32.  
    33. }
    • 生产者代码，生产者发送消息给交换机时指定routingKey 。
    1. import org.slf4j.Logger;
    2. import org.slf4j.LoggerFactory;
    3. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    4. import org.springframework.beans.factory.annotation.Autowired;
    5. import org.springframework.web.bind.annotation.GetMapping;
    6. import org.springframework.web.bind.annotation.RequestMapping;
    7. import org.springframework.web.bind.annotation.RestController;
    8. /**
    9.  * @author: gjm
    10.  * @date: 2021/9/05 19:14
    11.  * @description: 主题模式的生产者
    12.  */
    13. @RestController
    14. @RequestMapping("/producer")
    15. public class ProducerController {
    16.     private Logger logger = LoggerFactory.getLogger(ProducerController.class);
    17.     @Autowired
    18.     private RabbitTemplate rabbitTemplate;
    19.  
    20.     @RequestMapping("/topic")
    21.     public String sendTopicMessage(){
    22.         String message = "topic mode message";
    23.         rabbitTemplate.convertAndSend("topic_exchange","topic.test", message);
    24.         logger.info("主题模式的生产者发送了消息：{}", message);
    25.         return "success";
    26.     }
    27. }
    • 消费者代码，消费者端直接监听队列
    1. import com.rabbitmq.client.Channel;
    2. import org.slf4j.Logger;
    3. import org.slf4j.LoggerFactory;
    4. import org.springframework.amqp.core.Message;
    5. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    6. import org.springframework.stereotype.Component;
    7. import java.io.IOException;
    8. /**
    9.  * @author: gjm
    10.  * @date: 2021/9/05 19:15
    11.  * @description: 主题模式的消费者
    12.  */
    13. @Component
    14. public class TopicConsumer {
    15.     private Logger logger = LoggerFactory.getLogger(TopicConsumer.class);
    16.  
    17.     @RabbitListener(queues = "topic_queue")
    18.     public void consumer(String body, Message message, Channel channel) {
    19.         logger.info("主题模式的消费者接收了消息：{}", body);
    20.         try {
    21.             // 消费者手动确认消息已消费
    22.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    23.         } catch (IOException e) {
    24.             e.printStackTrace();
    25.         }
    26.     }
    27. }

     5.6 发布确认 

    • 配置文件中配置：spring.rabbitmq.publisher-confirm-type: correlated；默认值为none，禁用发布确认模式；还可以设置为simple。
    1.   spring:
    2.     rabbitmq:
    3.       publisher-confirm-type: correlated  # 消息发送给交换机时会触发回调
    4.       publisher-returns: true     # 交换机无法路由到队列的消息会触发回调变成回退消息

    • 实现交换机确认的回调接口RabbitTemplate.ConfirmCallback

      1. import org.slf4j.Logger;
      2. import org.slf4j.LoggerFactory;
      3. import org.springframework.amqp.rabbit.connection.CorrelationData;
      4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
      5. import org.springframework.beans.factory.annotation.Autowired;
      6. import org.springframework.stereotype.Component;
      7.  
      8. import javax.annotation.PostConstruct;
      9. /**
      10.  * @author: gjm
      11.  * @date: 2021/9/05 20:09
      12.  * @description: 消息发送到交换机的确认回调实现
      13.  */
      14. @Component
      15. public class MyConfirmCallback implements RabbitTemplate.ConfirmCallback {
      16.    
      17.     private Logger logger = LoggerFactory.getLogger(MyConfirmCallback.class);
      18.     @Autowired
      19.     private RabbitTemplate rabbitTemplate;
      20.  
      21.     @PostConstruct
      22.     public void init(){
      23.         rabbitTemplate.setConfirmCallback(this);
      24.     }
      25.  
      26.     @Override
      27.     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
      28.         String id = correlationData.getId();
      29.         if(ack) {
      30.             logger.info("交换机确认收到id:{}的消息", id);
      31.         }else {
      32.             // 这里需要对发送失败id的消息做消息处理，后续做消息补偿，应保证消息可通过id找回
      33.             logger.info("交换机未收到id:{}的消息, 原因是：{}", id, cause);
      34.         }
      35.     }
      36. }
    • 发布确认模式的配置类
    1. import org.springframework.amqp.core.*;
    2. import org.springframework.context.annotation.Bean;
    3. import org.springframework.context.annotation.Configuration;
    4. /**
    5.  * @author: gjm
    6.  * @date: 2021/9/05 20:17
    7.  * @description: 发布确认模式配置类
    8.  */
    9. @Configuration
    10. public class PublisherConfirmConfig {
    11.     // 交换机名称
    12.     public static final String PUBLISHER_CONFIRM_EXCHANGE = "publisher_confirm_exchange";
    13.     // 队列的名称
    14.     public static final String PUBLISHER_CONFIRM_QUEUE = "publisher_confirm_queue";
    15.     // 路由键的值
    16.     public static final String CONFIRM_ROUTING_KEY = "confirm.routing.key";
    17.  
    18.     // 创建direct类型的交换机
    19.     @Bean
    20.     public DirectExchange confirmExchange(){
    21.         return new DirectExchange(PUBLISHER_CONFIRM_EXCHANGE);
    22.     }
    23.     // 创建队列
    24.     @Bean
    25.     public Queue confirmQueue(){
    26.         return QueueBuilder.durable(PUBLISHER_CONFIRM_QUEUE).build();
    27.     }
    28.     // 队列confirmQueue和交换机通过routingKey绑定
    29.     @Bean
    30.     public Binding confirmQueueBindingExchange(Queue confirmQueue, DirectExchange confirmExchange){
    31.         return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(CONFIRM_ROUTING_KEY);
    32.     }
    33. }
    • 发布确认的生产者代码
    1. import com.mj.rabbitmq.config.PublisherConfirmConfig;
    2. import org.slf4j.Logger;
    3. import org.slf4j.LoggerFactory;
    4. import org.springframework.amqp.rabbit.connection.CorrelationData;
    5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    6. import org.springframework.beans.factory.annotation.Autowired;
    7. import org.springframework.web.bind.annotation.GetMapping;
    8. import org.springframework.web.bind.annotation.PathVariable;
    9. import org.springframework.web.bind.annotation.RequestMapping;
    10. import org.springframework.web.bind.annotation.RestController;
    11. import java.util.UUID;
    12. /**
    13.  * @author: gjm
    14.  * @date: 202/9/05 20:35
    15.  * @description: 发布确认模式发送消息的生产者
    16.  */
    17. @RestController
    18. @RequestMapping("/confirm")
    19. public class PublisherConfirmController {
    20.  
    21.     private Logger logger = LoggerFactory.getLogger(PublisherConfirmController.class);
    22.     @Autowired
    23.     private RabbitTemplate rabbitTemplate;
    24.  
    25.     @GetMapping("/sendMsg/{confirmMsg}")
    26.     public void sendConfirmMessage(@PathVariable("confirmMsg")String confirmMsg) {
    27.         // 生成消息id，为了消息不丢失，应该保存消息到数据库
    28.         String msgId = UUID.randomUUID().toString();
    29.         // 生成消息的唯一标识信息
    30.         CorrelationData correlationData = new CorrelationData(msgId);
    31.         rabbitTemplate.convertAndSend(PublisherConfirmConfig.PUBLISHER_CONFIRM_EXCHANGE,
    32.                 PublisherConfirmConfig.CONFIRM_ROUTING_KEY, confirmMsg, correlationData);
    33.         logger.info("发布确认的生产者发送消息：{}", confirmMsg);
    34.     }
    35. }
    •  发布确认的消费者代码
    1. import com.mj.rabbitmq.config.PublisherConfirmConfig;
    2. import com.rabbitmq.client.Channel;
    3. import org.slf4j.Logger;
    4. import org.slf4j.LoggerFactory;
    5. import org.springframework.amqp.core.Message;
    6. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    7. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    8. import org.springframework.stereotype.Component;
    9.  
    10. import java.io.IOException;
    11. /**
    12.  * @author: gjm
    13.  * @date: 2021/9/05 12:41
    14.  * @description: 发布确认模式发送消息的消费者
    15.  */
    16. @Component
    17. @RabbitListener(queues = PublisherConfirmConfig.PUBLISHER_CONFIRM_QUEUE)
    18. public class PublisherConfirmConsumer {
    19.  
    20.     private Logger logger = LoggerFactory.getLogger(PublisherConfirmConsumer.class);
    21.  
    22.     @RabbitHandler
    23.     public void receive(String msg, Message message, Channel channel){
    24.         logger.info("发布确认的消费者收到了消息：{}", msg);
    25.         try {
    26.             // 消费者手动确认消息已消费
    27.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    28.         } catch (IOException e) {
    29.             e.printStackTrace();
    30.         }
    31.     }
    32. }
    • 查看消息发送给交换机后的确认回调结果

    • 实现交换机无法路由到队列的消息进行回调的接口RabbitTemplate.ReturnCallback 
    1. import org.slf4j.Logger;
    2. import org.slf4j.LoggerFactory;
    3. import org.springframework.amqp.core.Message;
    4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    5. import org.springframework.beans.factory.annotation.Autowired;
    6. import org.springframework.stereotype.Component;
    7. import javax.annotation.PostConstruct;
    8. /**
    9.  * @author: gjm
    10.  * @date: 2021/09/05 21:02
    11.  * @description: 消息发送到交换机的确认回调实现
    12.  */
    13. @Component
    14. public class MyReturnCallback implements RabbitTemplate.ReturnCallback {
    15.  
    16.     private Logger logger = LoggerFactory.getLogger(MyReturnCallback.class);
    17.     @Autowired
    18.     private RabbitTemplate rabbitTemplate;
    19.  
    20.     @PostConstruct
    21.     public void init() {
    22.         rabbitTemplate.setReturnCallback(this);
    23.     }
    24.     @Override
    25.     public void returnedMessage(Message message, int replyCode, String replyText,
    26.                                 String exchange, String routingKey) {
    27.         logger.info("replyCode：{} -- replyText：{} -- exchange：{} -- routingKey：{}",
    28.                 replyCode, replyText, exchange, routingKey);
    29.         logger.info("交换机无法路由到队列的消息：{}", new String(message.getBody()));
    30.  
    31.     }
    32. }
    •  产生回退消息的生产者代码
    1. import com.mj.rabbitmq.config.PublisherConfirmConfig;
    2. import org.slf4j.Logger;
    3. import org.slf4j.LoggerFactory;
    4. import org.springframework.amqp.rabbit.connection.CorrelationData;
    5. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    6. import org.springframework.beans.factory.annotation.Autowired;
    7. import org.springframework.web.bind.annotation.GetMapping;
    8. import org.springframework.web.bind.annotation.PathVariable;
    9. import org.springframework.web.bind.annotation.RequestMapping;
    10. import org.springframework.web.bind.annotation.RestController;
    11. import java.util.UUID;
    12. /**
    13.  * @author: gjm
    14.  * @date: 2021/9/05 21:05
    15.  * @description: 发布确认发送消息的生产者
    16.  */
    17. @RestController
    18. @RequestMapping("/confirm")
    19. public class PublisherConfirmController {
    20.  
    21.     private Logger logger = LoggerFactory.getLogger(PublisherConfirmController.class);
    22.     @Autowired
    23.     private RabbitTemplate rabbitTemplate;
    24.  
    25.     @GetMapping("/sendReturnMsg/{returnMsg}")
    26.     public void sendReturnMessage(@PathVariable("returnMsg")String returnMsg) {
    27.         // 生成消息id，为了消息不丢失，应该保存消息到数据库
    28.         String msgId = UUID.randomUUID().toString();
    29.         // 生成消息的唯一标识信息
    30.         CorrelationData correlationData = new CorrelationData(msgId);
    31.         rabbitTemplate.convertAndSend(PublisherConfirmConfig.PUBLISHER_CONFIRM_EXCHANGE,
    32.                  "return_routing_key", returnMsg, correlationData);
    33.         logger.info("发布确认的生产者发送消息：{}", returnMsg);
    34.     }
    35. }
    • 生产者发送消息后查看结果，交换机可以接收到消息，队列无法接收到消息 ，从回调函数中的打印结果中我们知道消息无法路由的原因是routingKey不存在，消息变成回退消息。

  • 相关阅读:
    Leetcode 349.两个数组的交集
    MySQL的全局锁和表级锁
    HTML - input type=file 允许用户选择多个文件
    Kafka的docker安装
    Linux实用操作-----软件的安装
    电商商品详情 API（商品主图、sku）
    《智能风控实践指南》笔记（二）
    1796. 字符串中第二大的数字
    【每日练习】删除公共字符
    APP信息侦察&夜神模拟器Burp抓包配置
  • 原文地址：https://blog.csdn.net/G_12_M/article/details/127635172