• RabbitMQ学习总结

    目录

    一:介绍        

    二:应用场景

    三:工作原理

    组成部分说明

    消息发布接收流程

    四:下载安装

    五:环境搭建

    创建Maven工程

    生产者

    消费者 

     六:工作模式

    Work queues

    Publish/subscribe

    生产者

    消费者 

    Routing 

    生产者

    ​编辑消费者

     思考

     Topics

    生产者

    匹配规则 

    Header模式

    生产者

    RPC

    说明

    七:SpringBoot整合RabbitMQ

    一:添加依赖

    二:配置文件

    三:配置类

    四:生产者

    五:消费者 

    八:高级特性

    消费者绑定信息

    消息可靠性投递

    产生原因

    解决方案

    代码实现

    生产端确认

     备份交换机

     ACK/NACK机制

    总结 

    一:介绍        

    MQ全称为Message Queue，即消息队列， RabbitMQ是由erlang语言开发.

    RabbitMQ官方地址：http://www.rabbitmq.com/

    二:应用场景

    1)任务异步处理

    2)应用程序解耦合

    三:工作原理

    组成部分说明

     Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue。

     Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。

     Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。

     Producer：消息生产者，即生产方客户端，生产方客户端将消息发送到MQ。

     Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。

    消息发布接收流程

     发送消息:

     1、生产者和Broker建立TCP连接。
     2、生产者和Broker建立通道。
     3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发。
     4、Exchange将消息转发到指定的Queue（队列）

     接收消息:

     1、消费者和Broker建立TCP连接
     2、消费者和Broker建立通道
     3、消费者监听指定的Queue（队列）
     4、当有消息到达Queue时Broker默认将消息推送给消费者。
     5、消费者接收到消息

    四:下载安装

     RabbitMQ的下载地址：http://www.rabbitmq.com/download.html

     本项目使用Erlang/OTP 20.3版本和RabbitMQ3.7.3版本。

     下载erlang: http://erlang.org/download/otp_win64_20.3.exe

      erlang安装完成需要配置erlang环境变量： ERLANG_HOME=D:\Program Files\erl9.3 在path中 添加%ERLANG_HOME%\bin;

    安装erlang直接下一步就好

     安装RabbitMQ: Release RabbitMQ 3.7.3 · rabbitmq/rabbitmq-server · GitHub

    安装完开始菜单会有显示:

    RabbitMQ Service-install :安装服务
    RabbitMQ Service-remove 删除服务
    RabbitMQ Service-start 启动
    RabbitMQ Service-stop 启动

    如果没有开始菜单则进入安装目录下sbin目录手动启动：

    1）安装并运行服务
    rabbitmq-service.bat install 安装服务 rabbitmq-service.bat stop 停止服务 rabbitmq-service.bat start 启动服务
    2）安装管理插件
    安装rabbitMQ的管理插件，方便在浏览器端管理RabbitMQ
    管理员身份运行 rabbitmq-plugins.bat enable rabbitmq_management

    3、启动成功 登录RabbitMQ
    进入浏览器，输入：http://localhost:15672

    初始账号和密码：guest/guest

    注:每个虚拟机就相当于一个独立的MQ,默认虚拟机的名字为/

    4、如果启动失败找到.erlang.cookie,位于C:\windows\system32\config\systemprofile下，将此处的.erlang.cookie覆盖C:\user\admin.erlang.cookie后重启RabbitMQ即可解决问题

    注意事项

    1、安装erlang和rabbitMQ以管理员身份运行。
    2、当卸载重新安装时会出现RabbitMQ服务注册失败，此时需要进入注册表清理erlang搜索RabbitMQ、ErlSrv，将对应的项全部删除。

    五:环境搭建

    创建Maven工程

    创建生产者工程和消费者工程，分别加入RabbitMQ java client的依赖。
    test-rabbitmq-producer：生产者工程
    test-rabbitmq-consumer：消费者工程


    com.rabbitmq
    amqp‐client
    4.0.3


    org.springframework.boot
    spring‐boot‐starter‐logging

    生产者

    1. public class Producer01 {
    2.     //队列名称
    3.     private static final String QUEUE = "helloworld";
    4.  
    5.     public static void main(String[] args) throws IOException, TimeoutException {
    6.         Connection connection = null;
    7.         Channel channel = null;
    8.         try {
    9.             ConnectionFactory factory = new ConnectionFactory();
    10.             factory.setHost("localhost");
    11.             factory.setPort(5672);
    12.             factory.setUsername("guest");
    13.             factory.setPassword("guest");
    14.             factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
    15.             //创建与RabbitMQ服务的TCP连接
    16.             connection = factory.newConnection();
    17.             //创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
    18.             channel = connection.createChannel();
    19.  
    20.             /**
    21.              * 声明队列，如果Rabbit中没有此队列将自动创建
    22.              * param1:队列名称
    23.              * param2:durable是否持久化,如果持久化,mq重启后队列还在
    24.              * param3:exclusive队列是否独占此连接,队列只允许在该连接中访问,如果连接关闭队列自动删除,如果将此参数设置为true可用于临时队列的创建
    25.              * param4:autoDelete队列不再使用时是否自动删除此队列和exclusive搭配使用
    26.              * param5:队列参数(可以设置队列的扩展参数)
    27.              */
    28.             channel.queueDeclare(QUEUE, true, false, false, null);
    29.             String message = "helloworld小明" + System.currentTimeMillis();
    30.             /**
    31.              * 消息发布方法
    32.              * param1：Exchange的名称，如果没有指定，则使用Default Exchange
    33.              * param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列,如果使用默认交换机,routingKey设置为队列的名称
    34.              * param3:消息包含的属性
    35.              * param4：消息体
    36.              */
    37.  
    38.             /**
    39.              * 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显
    40.              示绑定或解除绑定
    41.              * 默认的交换机，routingKey等于队列名称
    42.              */
    43.             channel.basicPublish("", QUEUE, null, message.getBytes());
    44.             System.out.println("Send Message is:'" + message + "'");
    45.         } catch (Exception ex) {
    46.             ex.printStackTrace();
    47.         } finally {
    48.             if (channel != null) {
    49.                 channel.close();
    50.             }
    51.             if (connection != null) {
    52.                 connection.close();
    53.             }
    54.         }
    55.     }
    56. }

    此时显示创建了一个队列,有1条消息待消费, 消息总数为1

     

     注:队列中是可以拿到发送的消息的

     注:Purge Message可以清空该队列的消息

     注:Publish message可以在指定的队列中发消息,因此生产者发送消息时要打印发送的body,消费者端做幂等,一旦出现问题可通过控制台重新发送该消息

    消费者 

    注:两边都要声明队列,防止消费者启动在生产者之前报错

    1. public class Consumer01 {
    2.     private static final String QUEUE = "helloworld";
    3.  
    4.     public static void main(String[] args) throws IOException, TimeoutException {
    5.         ConnectionFactory factory = new ConnectionFactory();
    6.         //设置MabbitMQ所在服务器的ip和端口
    7.         factory.setHost("127.0.0.1");
    8.         factory.setPort(5672);
    9.         Connection connection = factory.newConnection();
    10.         Channel channel = connection.createChannel();
    11.         //声明队列
    12.         channel.queueDeclare(QUEUE, true, false, false, null);
    13.         //定义消费方法
    14.         DefaultConsumer consumer = new DefaultConsumer(channel) {
    15.             /**
    16.              * 消费者接收消息调用此方法
    17.              * @param consumerTag 消费者的标签，在channel.basicConsume()去指定
    18.              * @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
    19.             (收到消息失败后是否需要重新发送)
    20.              * @param properties
    21.              * @param body
    22.              * @throws IOException
    23.             }
    24.              */
    25.             @Override
    26.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    27.                 //交换机
    28.                 String exchange = envelope.getExchange();
    29.                 //路由key
    30.                 String routingKey = envelope.getRoutingKey();
    31.                 //消息id
    32.                 long deliveryTag = envelope.getDeliveryTag();
    33.                 //消息内容
    34.                 String msg = new String(body);
    35.                 System.out.println("receive message.." + msg);
    36.             }
    37.         };
    38.         /**
    39.          * 监听队列String queue, boolean autoAck,Consumer callback
    40.          * 参数明细
    41.          * 1、队列名称
    42.          * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
    43.          为false则需要手动回复
    44.          * 3、消费消息的方法，消费者接收到消息后调用此方法
    45.          */
    46.         channel.basicConsume(QUEUE, true, consumer);
    47.     }
    48. }

     六:工作模式

    RabbitMQ有以下几种工作模式 ：
    1、Work queues
    2、Publish/Subscribe
    3、Routing
    4、Topics
    5、Header
    6、RPC

    Work queues

    工作队列模式

    work queues与入门程序相比:多了一个消费端，两个消费端共同消费同一个队列中的消息。
    应用场景：对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

    测试：
    1、使用入门程序，启动多个消费者。
    2、生产者发送多个消息。
    结果：
    1、一条消息只会被一个消费者接收；
    2、rabbit采用轮询的方式将消息是平均发送给消费者的；
    3、消费者在处理完某条消息后，才会收到下一条消息。

     设置idea同时启用多个客户端

    Publish/subscribe

    发布订阅模式

    发布订阅模式：
    1、每个消费者监听自己的队列。
    2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息 

    案例：
    用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法 。

    生产者

    1、声明exchange_fanout_inform交换机。
    2、声明两个队列并且绑定到此交换机，绑定时不需要指定routingkey
    3、发送消息时不需要指定routingkey

    1. public class Producer02_publish {
    2.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    3.     private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    4.     private static final String EXCHANGE_FANOUT_INFORM = "exchange_fanout_inform";
    5.  
    6.     public static void main(String[] args) {
    7.         Connection connection = null;
    8.         Channel channel = null;
    9.         try {
    10.             //创建一个与MQ的连接
    11.             ConnectionFactory factory = new ConnectionFactory();
    12.             factory.setHost("127.0.0.1");
    13.             factory.setPort(5672);
    14.             factory.setUsername("guest");
    15.             factory.setPassword("guest");
    16.             factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
    17.             //创建一个连接
    18.             connection = factory.newConnection();
    19.             //创建与交换机的通道，每个通道代表一个会话
    20.             channel = connection.createChannel();
    21.             //声明交换机 String exchange, BuiltinExchangeType type
    22.             /**
    23.              * 参数明细
    24.              * 1、交换机名称
    25.              * 2、交换机类型，fanout、topic、direct、headers
    26.              */
    27.             channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, "fanout");
    28.             //声明队列
    29.             // (String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
    30.             /**
    31.              * 参数明细：
    32.              * 1、队列名称
    33.              * 2、是否持久化
    34.              * 3、是否独占此队列
    35.              * 4、队列不用是否自动删除
    36.              * 5、参数
    37.              */
    38.             channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
    39.             channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null);
    40.             //交换机和队列绑定String queue, String exchange, String routingKey
    41.             /**
    42.              * 参数明细
    43.              * 1、队列名称
    44.              * 2、交换机名称
    45.              * 3、路由key
    46.              */
    47.             channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, "");
    48.             channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_FANOUT_INFORM, "");
    49.             //发送消息
    50.             for (int i = 0; i < 10; i++) {
    51.                 String message = "inform to user" + i;
    52.                 //向交换机发送消息 String exchange, String routingKey, BasicProperties props, byte[] body
    53.                 /**
    54.                  * 参数明细
    55.                  * 1、交换机名称，不指令使用默认交换机名称 Default Exchange
    56.                  * 2、routingKey（路由key），根据key名称将消息转发到具体的队列，这里填写队列名称表示消息将发到此队列
    57.                  * 3、消息属性
    58.                  * 4、消息内容
    59.                  */
    60.                 channel.basicPublish(EXCHANGE_FANOUT_INFORM, "", null, message.getBytes());
    61.                 System.out.println("Send Message is:'" + message + "'");
    62.             }
    63.         } catch (IOException e) {
    64.             e.printStackTrace();
    65.         } catch (TimeoutException e) {
    66.             e.printStackTrace();
    67.         } finally {
    68.             if (channel != null) {
    69.                 try {
    70.                     channel.close();
    71.                 } catch (IOException e) {
    72.                     e.printStackTrace();
    73.                 } catch (TimeoutException e) {
    74.                     e.printStackTrace();
    75.                 }
    76.             }
    77.             if (connection != null) {
    78.                 try {
    79.                     connection.close();
    80.                 } catch (IOException e) {
    81.                     e.printStackTrace();
    82.                 }
    83.             }
    84.         }
    85.     }
    86. }

     可以看到新声明的交换机exchange_fanout_inform

     新生成两个队列,每个队列接收到10条消息

    消费者 

    邮件发送消费者

    1. public class Consumer02_subscribe_email {
    2.     //队列名称
    3.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    4.     private static final String EXCHANGE_FANOUT_INFORM = "inform_exchange_fanout";
    5.  
    6.     public static void main(String[] args) throws IOException, TimeoutException {
    7.         //创建一个与MQ的连接
    8.         ConnectionFactory factory = new ConnectionFactory();
    9.         factory.setHost("127.0.0.1");
    10.         factory.setPort(5672);
    11.         factory.setUsername("guest");
    12.         factory.setPassword("guest");
    13.         factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
    14.         //创建一个连接
    15.         Connection connection = factory.newConnection();
    16.         //创建与交换机的通道，每个通道代表一个会话
    17.         Channel channel = connection.createChannel();
    18.         //声明交换机 String exchange, BuiltinExchangeType type
    19.         /**
    20.          * 参数明细
    21.          * 1、交换机名称
    22.          * 2、交换机类型，fanout、topic、direct、headers
    23.          */
    24.         channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, "fanout");
    25.         //声明队列
    26.         // channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
    27.         /**
    28.          * 参数明细：
    29.          * 1、队列名称
    30.          * 2、是否持久化
    31.          * 3、是否独占此队列
    32.          * 4、队列不用是否自动删除
    33.          * 5、参数
    34.          */
    35.         channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
    36.         //交换机和队列绑定String queue, String exchange, String routingKey
    37.         /**
    38.          * 参数明细
    39.          * 1、队列名称
    40.          * 2、交换机名称
    41.          * 3、路由key
    42.          */
    43.         channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, "");
    44.         //定义消费方法
    45.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
    46.             @Override
    47.             public void handleDelivery(String consumerTag, Envelope envelope,
    48.                                        AMQP.BasicProperties properties, byte[] body) throws IOException {
    49.                 long deliveryTag = envelope.getDeliveryTag();
    50.                 String exchange = envelope.getExchange();
    51.                 //消息内容
    52.                 String message = new String(body);
    53.                 System.out.println(message);
    54.             }
    55.         };
    56.         /**
    57.          * 监听队列String queue, boolean autoAck,Consumer callback
    58.          * 参数明细
    59.          * 1、队列名称
    60.          * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
    61.          为false则需要手动回复
    62.          * 3、消费消息的方法，消费者接收到消息后调用此方法
    63.          */
    64.         channel.basicConsume(QUEUE_INFORM_EMAIL, true, defaultConsumer);
    65.     }
    66.  
    67. }

    可以看到邮件的信息被消费了,短信的还在

    可以看到该交换机绑定了两个队列 

    Routing 

    路由模式：
    1、每个消费者监听自己的队列，并且设置routingkey。
    2、生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列。     

    生产者

    1、声明exchange_routing_inform交换机。
    2、声明两个队列并且绑定到此交换机，绑定时需要指定routingkey
    3、发送消息时需要指定routingkey

    1. public class Producer03_routing {
    2.     //队列名称
    3.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    4.     private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    5.     private static final String EXCHANGE_ROUTING_INFORM = "exchange_routing_inform";
    6.  
    7.     public static void main(String[] args) {
    8.         Connection connection = null;
    9.         Channel channel = null;
    10.         try {
    11.             //创建一个与MQ的连接
    12.             ConnectionFactory factory = new ConnectionFactory();
    13.             factory.setHost("127.0.0.1");
    14.             factory.setPort(5672);
    15.             factory.setUsername("guest");
    16.             factory.setPassword("guest");
    17.             factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
    18.             //创建一个连接
    19.             connection = factory.newConnection();
    20.             //创建与交换机的通道，每个通道代表一个会话
    21.             channel = connection.createChannel();
    22.             //声明交换机 String exchange, BuiltinExchangeType type
    23.             /**
    24.              * 参数明细
    25.              * 1、交换机名称
    26.              * 2、交换机类型，fanout、topic、direct、headers
    27.              */
    28.             channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, "direct");
    29.             //声明队列
    30.             // channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
    31.             /**
    32.              * 参数明细：
    33.              * 1、队列名称
    34.              * 2、是否持久化
    35.              * 3、是否独占此队列
    36.              * 4、队列不用是否自动删除
    37.              * 5、参数
    38.              */
    39.             channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
    40.             channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null);
    41.             //交换机和队列绑定String queue, String exchange, String routingKey
    42.             /**
    43.              * 参数明细
    44.              * 1、队列名称
    45.              * 2、交换机名称
    46.              * 3、路由key
    47.              */
    48.             channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_EMAIL);
    49.             channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_SMS);
    50.             //发送邮件消息
    51.             for (int i = 0; i < 10; i++) {
    52.                 String message = "email inform to user" + i;
    53.                 //向交换机发送消息 String exchange, String routingKey, BasicProperties props, byte[] body
    54.                 /**
    55.                  * 参数明细
    56.                  * 1、交换机名称，不指令使用默认交换机名称 Default Exchange
    57.                  * 2、routingKey（路由key），根据key名称将消息转发到具体的队列，这里填写队列名称表示消
    58.                  息将发到此队列
    59.                  * 3、消息属性
    60.                  * 4、消息内容
    61.                  */
    62.                 channel.basicPublish(EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_EMAIL, null, message.getBytes());
    63.                 System.out.println("Send Message is:'" + message + "'");
    64.             }
    65.             //发送短信消息
    66.             for (int i = 0; i < 10; i++) {
    67.                 String message = "sms inform to user" + i;
    68.                 //向交换机发送消息 String exchange, String routingKey, BasicProperties props, byte[] body
    69.                 channel.basicPublish(EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_SMS, null, message.getBytes());
    70.                 System.out.println("Send Message is:'" + message + "'");
    71.             }
    72.         } catch (IOException e) {
    73.             e.printStackTrace();
    74.         } catch (TimeoutException e) {
    75.             e.printStackTrace();
    76.         } finally {
    77.             if (channel != null) {
    78.                 try {
    79.                     channel.close();
    80.                 } catch (IOException e) {
    81.                     e.printStackTrace();
    82.                 } catch (TimeoutException e) {
    83.                     e.printStackTrace();
    84.                 }
    85.             }
    86.             if (connection != null) {
    87.                 try {
    88.                     connection.close();
    89.                 } catch (IOException e) {
    90.                     e.printStackTrace();
    91.                 }
    92.             }
    93.         }
    94.     }
    95. }

    可以看到创建了exchange_routing_inform交换机,并绑定了两个队列

    可以看到对列绑定了Routing_key

    消费者

    邮件发送消费者

    1. public class Consumer03_routing_email {
    2.     //队列名称
    3.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    4.     private static final String EXCHANGE_ROUTING_INFORM = "exchange_routing_inform";
    5.  
    6.     public static void main(String[] args) throws IOException, TimeoutException {
    7.         //创建一个与MQ的连接
    8.         ConnectionFactory factory = new ConnectionFactory();
    9.         factory.setHost("127.0.0.1");
    10.         factory.setPort(5672);
    11.         factory.setUsername("guest");
    12.         factory.setPassword("guest");
    13.         factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
    14.         //创建一个连接
    15.         Connection connection = factory.newConnection();
    16.         //创建与交换机的通道，每个通道代表一个会话
    17.         Channel channel = connection.createChannel();
    18.         //声明交换机 String exchange, BuiltinExchangeType type
    19.         /**
    20.          * 参数明细
    21.          * 1、交换机名称
    22.          * 2、交换机类型，fanout、topic、direct、headers
    23.          */
    24.         channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, "direct");
    25.         //声明队列
    26.         // channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
    27.         /**
    28.          * 参数明细：
    29.          * 1、队列名称
    30.          * 2、是否持久化
    31.          * 3、是否独占此队列
    32.          * 4、队列不用是否自动删除
    33.          * 5、参数
    34.          */
    35.         channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
    36.         //交换机和队列绑定String queue, String exchange, String routingKey
    37.         /**
    38.          * 参数明细
    39.          * 1、队列名称
    40.          * 2、交换机名称
    41.          * 3、路由key
    42.          */
    43.         channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_EMAIL);
    44.         //定义消费方法
    45.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
    46.             @Override
    47.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    48.                 long deliveryTag = envelope.getDeliveryTag();
    49.                 String exchange = envelope.getExchange();
    50.                 //消息内容
    51.                 String message = new String(body);
    52.                 System.out.println(message);
    53.             }
    54.         };
    55.         /**
    56.          * 监听队列String queue, boolean autoAck,Consumer callback
    57.          * 参数明细
    58.          * 1、队列名称
    59.          * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
    60.          为false则需要手动回复
    61.          * 3、消费消息的方法，消费者接收到消息后调用此方法
    62.          */
    63.         channel.basicConsume(QUEUE_INFORM_EMAIL, true, defaultConsumer);
    64.     }
    65.  
    66. }

     思考

    1、Routing模式和Publish/subscibe有啥区别？
    Routing模式要求队列在绑定交换机时要指定routingkey，消息会转发到符合routingkey的队列。

     Topics

    路由模式：
    1、每个消费者监听自己的队列，并且设置带统配符的routingkey。
    2、生产者将消息发给broker，由交换机根据routingkey来转发消息到指定的队列。 

    案例：

    根据用户的通知设置去通知用户，设置接收Email的用户只接收Email，设置接收sms的用户只接收sms，设置两种通知类型都接收的则两种通知都有效。

    生产者

    1. public class Producer04_topics {
    2.     //队列名称
    3.     private static final String QUEUE_INFORM_EMAIL = "queue_inform_email_topics";
    4.     private static final String QUEUE_INFORM_SMS = "queue_inform_sms_topics";
    5.     private static final String EXCHANGE_TOPICS_INFORM = "exchange_topics_inform";
    6.  
    7.     public static void main(String[] args) {
    8.         Connection connection = null;
    9.         Channel channel = null;
    10.         try {
    11.             //创建一个与MQ的连接
    12.             ConnectionFactory factory = new ConnectionFactory();
    13.             factory.setHost("127.0.0.1");
    14.             factory.setPort(5672);
    15.             factory.setUsername("guest");
    16.             factory.setPassword("guest");
    17.             factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器
    18.             //创建一个连接
    19.             connection = factory.newConnection();
    20.             //创建与交换机的通道，每个通道代表一个会话
    21.             channel = connection.createChannel();
    22.             //声明交换机 String exchange, BuiltinExchangeType type
    23.             /**
    24.              * 参数明细
    25.              * 1、交换机名称
    26.              * 2、交换机类型，fanout、topic、direct、headers
    27.              */
    28.             channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, "topic");
    29.             //声明队列
    30.             /**
    31.              * 参数明细：
    32.              * 1、队列名称
    33.              * 2、是否持久化
    34.              * 3、是否独占此队列
    35.              * 4、队列不用是否自动删除
    36.              * 5、参数
    37.              */
    38.             channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
    39.             channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null);
    40.  
    41.             //绑定email通知队列
    42.             channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_TOPICS_INFORM,"inform.#.email.#");
    43.             //绑定sms通知队列
    44.             channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_TOPICS_INFORM,"inform.#.sms.#");
    45.  
    46.             //发送邮件消息
    47.             for (int i = 0; i < 10; i++) {
    48.                 String message = "email inform to user" + i;
    49.                 //向交换机发送消息 String exchange, String routingKey, BasicProperties props, byte[] body
    50.                 /**
    51.                  * 参数明细
    52.                  * 1、交换机名称，不指令使用默认交换机名称 Default Exchange
    53.                  * 2、routingKey（路由key），根据key名称将消息转发到具体的队列，这里填写队列名称表示消
    54.                  息将发到此队列
    55.                  * 3、消息属性
    56.                  * 4、消息内容
    57.                  */
    58.                 channel.basicPublish(EXCHANGE_TOPICS_INFORM, "inform.email", null, message.getBytes());
    59.                 System.out.println("Send Message is:'" + message + "'");
    60.             }
    61.             //发送短信消息
    62.             for (int i = 0; i < 10; i++) {
    63.                 String message = "sms inform to user" + i;
    64.                 channel.basicPublish(EXCHANGE_TOPICS_INFORM, "inform.sms", null, message.getBytes());
    65.                 System.out.println("Send Message is:'" + message + "'");
    66.             }
    67.             //发送短信和邮件消息
    68.             for (int i = 0; i < 10; i++) {
    69.                 String message = "sms and email inform to user" + i;
    70.                 channel.basicPublish(EXCHANGE_TOPICS_INFORM, "inform.sms.email", null, message.getBytes());
    71.                 System.out.println("Send Message is:'" + message + "'");
    72.             }
    73.         } catch (IOException | TimeoutException e) {
    74.             e.printStackTrace();
    75.         } finally {
    76.             if (channel != null) {
    77.                 try {
    78.                     channel.close();
    79.                 } catch (IOException e) {
    80.                     e.printStackTrace();
    81.                 } catch (TimeoutException e) {
    82.                     e.printStackTrace();
    83.                 }
    84.             }
    85.             if (connection != null) {
    86.                 try {
    87.                     connection.close();
    88.                 } catch (IOException e) {
    89.                     e.printStackTrace();
    90.                 }
    91.             }
    92.         }
    93.     }
    94.  
    95. }

    可以看到exchange_topics_inform绑定了两个Routing_key,进行通配符匹配

    匹配规则 

    统配符规则：
    中间以“.”分隔。
    符号#可以匹配多个词，符号*可以匹配一个词语。

    对列绑定的Routing_key为:

    inform.#.email.#

    inform.#.sms.#

    发送消息指定的Routing_key为:

    inform.email 匹配 inform.#.email.#

    inform.sms 匹配 inform.#.sms.#

    inform.sms.email 匹配 inform.#.email.# 和 inform.#.sms.# 两个队列都能发送

    (即使什么都没有,#也可以匹配个空)

    Header模式

    header模式与routing不同的地方在于，header模式取消routingkey，使用header中的 key/value（键值对）匹配队列。

    案例：
    根据用户的通知设置去通知用户，设置接收Email的用户只接收Email，设置接收sms的用户只接收sms，设置两种通知类型都接收的则两种通知都有效

    生产者

    1. Map headers_email = new Hashtable();
    2. headers_email.put("inform_type", "email");
    3. Map headers_sms = new Hashtable();
    4. headers_sms.put("inform_type", "sms");
    5. channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);
    6. channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_HEADERS_INFORM,"",headers_sms);

    通知

    1. String message = "email inform to user"+i;
    2. Map headers = new Hashtable();
    3. headers.put("inform_type", "email");//匹配email通知消费者绑定的header
    4. //headers.put("inform_type", "sms");//匹配sms通知消费者绑定的header
    5. AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties.Builder();
    6. properties.headers(headers);
    7. //Email通知
    8. channel.basicPublish(EXCHANGE_HEADERS_INFORM, "", properties.build(), message.getBytes());

    发送邮件消费者 

    1. channel.exchangeDeclare(EXCHANGE_HEADERS_INFORM, BuiltinExchangeType.HEADERS);
    2. Map headers_email = new Hashtable();
    3. headers_email.put("inform_email", "email");
    4. //交换机和队列绑定
    5. channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);
    6. //指定消费队列
    7. channel.basicConsume(QUEUE_INFORM_EMAIL, true, consumer);

    RPC

    说明

    RPC即客户端远程调用服务端的方法 ，使用MQ可以实现RPC的异步调用，基于Direct交换机实现，流程如下：
    1、客户端即是生产者就是消费者，向RPC请求队列发送RPC调用消息，同时监听RPC响应队列。
    2、服务端监听RPC请求队列的消息，收到消息后执行服务端的方法，得到方法返回的结果
    3、服务端将RPC方法 的结果发送到RPC响应队列
    4、客户端（RPC调用方）监听RPC响应队列，接收到RPC调用结果。

    七:SpringBoot整合RabbitMQ

    一:添加依赖

    2.    org.springframework.boot
    3.    spring-boot-starter-amqp
    4.    3.3.0

    二:配置文件

    1. server:
    2.   port: 44000
    3. spring:
    4.   application:
    5.     name: test-SpringBoot-RabbitMQ-product
    6.   rabbitmq:
    7.     host: 127.0.0.1
    8.     port: 5672
    9.     username: guest
    10.     password: guest
    11.     virtualHost: /

    三:配置类

    1. @Configuration
    2. public class RabbitmqConfig {
    3.     public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    4.     public static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    5.     public static final String EXCHANGE_TOPICS_INFORM = "exchange_topics_inform";
    6.  
    7.     /**
    8.      * 交换机配置
    9.      * ExchangeBuilder提供了fanout、direct、topic、header交换机类型的配置
    10.      *
    11.      * @return the exchange
    12.      */
    13.     @Bean(EXCHANGE_TOPICS_INFORM)
    14.     public Exchange EXCHANGE_TOPICS_INFORM() {
    15.         //durable(true)持久化，消息队列重启后交换机仍然存在
    16.         return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build();
    17.     }
    18.  
    19.     //声明队列
    20.     @Bean(QUEUE_INFORM_SMS)
    21.     public Queue QUEUE_INFORM_SMS() {
    22.         Queue queue = new Queue(QUEUE_INFORM_SMS);
    23.         return queue;
    24.     }
    25.  
    26.     //声明队列
    27.     @Bean(QUEUE_INFORM_EMAIL)
    28.     public Queue QUEUE_INFORM_EMAIL() {
    29.         Queue queue = new Queue(QUEUE_INFORM_EMAIL);
    30.         return queue;
    31.     }
    32.  
    33.     /**
    34.      * channel.queueBind(INFORM_QUEUE_SMS,"inform_exchange_topic","inform.#.sms.#");
    35.      * 绑定队列到交换机 .
    36.      *
    37.      * @param queue    the queue
    38.      * @param exchange the exchange
    39.      * @return the binding
    40.      */
    41.     @Bean
    42.     public Binding BINDING_QUEUE_INFORM_SMS(@Qualifier(QUEUE_INFORM_SMS) Queue queue,
    43.                                             @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange) {
    44.         return BindingBuilder.bind(queue).to(exchange).with("inform.#.sms.#").noargs();
    45.     }
    46.  
    47.     @Bean
    48.     public Binding BINDING_QUEUE_INFORM_EMAIL(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue,
    49.                                               @Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange) {
    50.         return BindingBuilder.bind(queue).to(exchange).with("inform.#.email.#").noargs();
    51.     }
    52. }

    四:生产者

    1. @Test
    2.     public void testSendByTopics(){
    3.         for (int i=0;i<5;i++){
    4.             String message = "sms email inform to user"+i;
    5.             rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_TOPICS_INFORM,"inform.sms.email",message);
    6.             System.out.println("Send Message is:'" + message + "'");
    7.         }
    8.     }

    五:消费者 

    1. @Component
    2. public class ReceiveHandler {
    3.     public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
    4.     public static final String QUEUE_INFORM_SMS = "queue_inform_sms";
    5.     public static final String EXCHANGE_TOPICS_INFORM = "exchange_topics_inform";
    6.     //监听email队列
    7.     @RabbitListener(queues = {ReceiveHandler.QUEUE_INFORM_EMAIL})
    8.     public void receive_email(String msg, Message message, Channel channel){
    9.         System.out.println(msg);
    10.     }
    11.     //监听sms队列
    12.     @RabbitListener(queues = {ReceiveHandler.QUEUE_INFORM_SMS})
    13.     public void receive_sms(String msg,Message message,Channel channel){
    14.         System.out.println(msg);
    15.     }
    16. }

    八:高级特性

    消费者绑定信息

    1)通过@RabbitListener注解启动消费端也可以声明队列,交换机,路由信息...等等,如下:

    2)尽量不要这么写,如果前置的绑定关系已经创建,只需要 @RabbitListener(queues = {ReceiveHandler.QUEUE_INFORM_SMS})用这种写法(推荐)

    3)甚至直接在rabbitMQ图形化界面创建

    1. @Component
    2. public class ReceiveHandlerV2 {
    3.     public static final String ROUTING_KEY = "order";
    4.     public static final String QUEUE_NAME = "queue.order";
    5.     public static final String EXCHANGE_DIRECT = "exchange.direct.order";
    6.  
    7.  
    8.     //监听email队列
    9.     @RabbitListener(bindings = @QueueBinding(
    10.             value = @Queue(value = QUEUE_NAME, durable = "true"),
    11.             exchange = @Exchange(value = EXCHANGE_DIRECT),
    12.             key = {ROUTING_KEY}
    13.     ))
    14.     public void processMessage(String msg, Message message, Channel channel){
    15.         System.out.println(msg);
    16.     }
    17. }

    消息可靠性投递

    产生原因

    1)消息没有发送到消息对列上

    2)消息成功存入消息队列,但是消息队列服务器宕机,原本保存在内存中的消息丢失

    3)消息成功存入消息队列,但是消费端服务宕机,抛异常等等...

    后果:消费者拿不到消息,业务功能缺失,数据错误

    解决方案

    1) 消息没有发送到消息队列:

    • 解决思路A:在生产端进行确认,具体操作中我们会针对交换机和队列来确认,如果没有成功发送到消息队列服务器上,那就可以尝试重新发送
    • 解决思路B:为目标交换机指定备份交换机,当目标交换机投递失败时,把消息投递到备份交换机

    2) 宕机导致内存中消息丢失

    • 解决思路:消息持久化到硬盘

    3) 但是消费端服务宕机,抛异常

    • 消费端消费成功,给服务器返回ACK信息,然后消息队列删除该消息
    • 消费端消费失败,给服务器返回NACK信息,同时把消息恢复为待消费状态,这样就可以再次取回消息,重试一次(消费者支持幂等)

    代码实现

    生产端确认

    注意:必须加如下两个配置,否则不生效

    创建配置类:Why?首先我们需要声明一个回调函数来接受RabbitMQ服务器返回的确认信息

    方法名方法功能所属接口接口所属类
    confirm()确认消息是否发送到交换机ConfirmCallbackRabbitTemplate
    returnedMessage()确认消息是否发送到对列ReturnsCallbackRabbitTemplate
    • 然后就是对RabbitTemplate的功能进行增强,因为回调函数所在对象必须设置到RabbitTemplate
    • 原本RabbitTemplate对象并没有生产端消息确认功能,要给它设置对应的组件才可以
    • 而设置对应的组件需要调用RabbitTemplate对象下面的两个方法
    调用组件的方法所需对象类型
    setConfirmCallback()ConfirmCallback接口类型
    setReturnsCallback()ReturnsCallback接口类型

     配置类信息代码,如下:

    1. package org.example.testspringbootrabbitmqproduct;
    2.  
    3. import jakarta.annotation.PostConstruct;
    4. import org.springframework.amqp.core.ReturnedMessage;
    5. import org.springframework.amqp.rabbit.connection.CorrelationData;
    6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
    7. import org.springframework.beans.factory.annotation.Autowired;
    8. import org.springframework.context.annotation.Configuration;
    9.  
    10. @Configuration
    11. public class RabbitConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback{
    12.  
    13.     @Autowired
    14.     private RabbitTemplate rabbitTemplate;
    15.  
    16.     /**
    17.      * 只放在IOC中是不够的,需要设置在RabbitTemplate中才可以
    18.      */
    19.     @PostConstruct
    20.     public void initRabbitTemplate() {
    21.         rabbitTemplate.setConfirmCallback(this);
    22.         rabbitTemplate.setReturnsCallback(this);
    23.     }
    24.  
    25.     /**
    26.      * @param correlationData correlation data for the callback.
    27.      * @param ack true for ack, false for nack
    28.      * @param cause An optional cause, for nack, when available, otherwise null.
    29.      */
    30.     @Override
    31.     public void confirm(CorrelationData correlationData, boolean ack, String cause) {
    32.         // 消息发送到交换机成功或者失败都会回调这个方法
    33.         System.out.println(correlationData);//成功后返回 null
    34.         System.out.println(ack);//成功后返回 true
    35.         System.out.println(cause);//成功后返回 null
    36.     }
    37.  
    38.     /**
    39.      * @param returnedMessage the returned message and metadata.
    40.      */
    41.     @Override
    42.     public void returnedMessage(ReturnedMessage returnedMessage) {
    43.         // 只有发送到队列上失败才会回调该方法
    44.         System.out.println("消息主体: " + new String(returnedMessage.getMessage().getBody()));
    45.         System.out.println("应答码: " + returnedMessage.getReplyCode());
    46.         System.out.println("描述: " + returnedMessage.getReplyText());
    47.         System.out.println("消息使用的交换器是: " + returnedMessage.getExchange());
    48.         System.out.println("消息使用的路由键是: " + returnedMessage.getRoutingKey());
    49.  
    50. //        消息主体: 11
    51. //        应答码: 312
    52. //        描述: NO_ROUTE
    53. //        消息使用的交换器是: exchange.direct.order
    54. //        消息使用的路由键是: inform.sms.email1111
    55.     }
    56. }
     备份交换机

    原理图

    注意:备份交换机的类型必须是fanout(因为是不带路由键的)

    注意:备份交换机只能是创建的时候指定

     模拟发送一个不存在的routingKey,查看备份队列已经有一条消息

     

     ACK/NACK机制

    代码中开启机制的支持

    思考:更新购物车的微服务消费了消息返回的ACK确认信息,然后Broker删除了信息,进而导致更新库存,更新积分的功能拿不到消息----这种情况会发生吗?  

    答案:会的,因此需要广播到不同的队列中

    • 消费端代码
    1. package org.example.testspringbootrabbitmqconsumer;
    2.  
    3. import com.rabbitmq.client.Channel;
    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. @Component
    11. public class ReceiveHandlerV2 {
    12.     public static final String ROUTING_KEY = "order";
    13.     public static final String QUEUE_NAME = "queue.order";
    14.     public static final String EXCHANGE_DIRECT = "exchange.direct.order";
    15.  
    16.  
    17.     //监听队列
    18.     @RabbitListener(queues = {QUEUE_NAME})
    19.     public void processMessage(String msg, Message message, Channel channel) throws IOException {
    20.         try {
    21.             // 成功了返回ACK
    22.             System.out.println("消费端接收到了消息: " + msg);
    23.             /**
    24.              * deliveryTag: 交付标签(对列中每条消息的唯一标识)是以为64位整数
    25.              *              会从生产者带过来
    26.              *              提问: 如果交换机是Fanout模式,同一个消息广播到不同对列,deliveryTag会重复吗?
    27.              *                   不会,deliveryTag在Broker范围内唯一
    28.              * multiple: 为true时 指定某个deliveryTag,为ACK时全部删除,为NACK时全部重新投递
    29.              *            为false时,只做单个的处理
    30.              */
    31.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    32.         } catch (Exception e) {
    33.             // 失败了返回NACK
    34.             // 获取当前消息是否是重复投递
    35.             Boolean redelivered = message.getMessageProperties().getRedelivered();
    36.             if (redelivered) {
    37.                 // 就不放回对列了,说明此前重复过了
    38.                 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
    39.             } else {
    40.                 /**
    41.                  * requeue: 控制消息是否重新放回队列 true:重新放回,进行投递  false:不重新投递, broker会丢弃
    42.                  */
    43.                 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    44.             }
    45.             throw new RuntimeException(e);
    46.         }
    47.     }
    48. }
    •  ACK情况演示

     发送一条消息未执行basicAck()时,队列中消息还再存在

     

    执行完basicAck()后队列中消息消失

    • NACK情况演示

     

     第一次是false

     第二次是true

    速率显示重新投递

     分别向两个队列添加100条信息

    强制让一个队列进行NACK 

     

    启动消费者 ,可以看到NACK的队列一直处于Runing状态中,什么都不做的队列消费完了

    结论: 配置文件配置的 acknowledge-mode: manual #把消息确认模式改为手动确认,只是起一个开启功能的作用,并不是每个队列都需要手动ack,如果不写手动ack的代码会自动进行ack

     

    // 表示拒绝, 和basicNack基本一样,区别是 是否能控制批量操作 channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);  

     

    总结 

    三个部分的操作相互配合总体上实现消息的可靠性投递

  • 相关阅读:
    ITSS信息技术服务各级认证条件有哪些？
    htmx-使HTML更强大
    安装虚拟机找不到虚拟网啦1（eth1）不出现
    数字翻牌器
    Jenkins+vue发布项目
    计算机视觉中的可解释性分析
    Linux进程控制/进程终止
    『现学现忘』Git基础 — 24、Git中查看历史版本记录
    Docker从认识到实践再到底层原理（五）｜Docker镜像
    JSP协同办公eclipse定制开发mysql数据库BS模式java编程OA系统
  • 原文地址：https://blog.csdn.net/weixin_55157621/article/details/139667316