SpringBoot的RabbitMQ消息队列: 三、第二模式Work queues

本文主要是介绍SpringBoot的RabbitMQ消息队列: 三、第二模式Work queues,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

  上一节的两个工程,一个负责发送,一个负责接收,也就是一一对于的关系。

     只要消息发出了,接收者就处理;当接收效率较低时,就会出现接收者处理不过来,我们就可能会处理不过来,于是我们就可能多配置接受者。这个模式就是"Work queues",它的结构如下


    多个接收者,它们会出现什么情况呢?是否像大锅饭,有的人撑死,有的人饿死。这个通过例子验证。

一、再建一个接收者工程 HelloReceiving2


1、把HelloReceiver工程中的HelloRabbitConfig、HelloReceiver、logback.xml依次拷贝过去

2、修改application.properties为

[html]  view plain copy
  1. #服务器配置  
  2. spring.application.name=rabbitmq-hello-receiving  
  3. server.port=9092   
  4. #rabbitmq连接参数  
  5. spring.rabbitmq.host=localhost  
  6. spring.rabbitmq.port=5672  
  7. spring.rabbitmq.username=test  
  8. spring.rabbitmq.password=123456  

二、运行工程

1、在工程HelloSending所在文件夹打开cmd,运行mvn spring-boot:run

2、在工程HelloReceiving所在文件夹打开cmd,运行mvn spring-boot:run

3、在工程HelloReceiving2所在文件夹打开cmd,运行mvn spring-boot:run
4、在浏览器中输入http://localhost:9080/send/上帝1,http://localhost:9080/send/上帝2,http://localhost:9080/send/上帝3
观察两个Receiving的日志.

查看出不均衡吧,为了突出这个不公平,我们修改发送代码如下

[html]  view plain copy
  1. package com.example;  
  2.   
  3. import java.util.Date;  
  4.   
  5. import org.slf4j.Logger;  
  6. import org.slf4j.LoggerFactory;  
  7. import org.springframework.amqp.core.AmqpTemplate;  
  8. import org.springframework.beans.factory.annotation.Autowired;  
  9. import org.springframework.stereotype.Component;  
  10.   
  11. @Component  
  12. public class HelloSender {  
  13.   
  14.     protected static Logger logger=LoggerFactory.getLogger(HelloSender.class);   
  15.       
  16.     @Autowired  
  17.     private AmqpTemplate rabbitTemplate;  
  18.   
  19.     public String send(String name) {  
  20.         String context = "hello "+name+" --" + new Date();  
  21.         String sendStr;  
  22.         for(int i=1;i<=100;i++){  
  23.             sendStr="第["+i+"]个 hello "+name+" --" + new Date();   
  24.             logger.debug("HelloSender: " + sendStr);  
  25.             this.rabbitTemplate.convertAndSend("hello", sendStr);  
  26.         }  
  27.         return context;  
  28.     }  
  29. }  

再次http://localhost:9080/send/上帝,会发现更多的不公平。

三、Message acknowledgment 消息确认

1、默认情况下,RabbitMQ 会顺序的分发每个Message。当分发后,会将该Message删除,然后将下一个Message分发到下一个Consumer。这种分发方式叫做round-robin

2、每个Consumer可能需要一段时间才能处理完收到的数据。如果在这个过程中,Consumer出错了,异常退出了,而数据还没有处理完成,那么非常不幸,这段数据就丢失了。因为我们采用no-ack的方式进行确认,也就是说,每次Consumer接到数据后,而不管是否处理完成,RabbitMQ Server会立即把这个Message标记为完成,然后从queue中删除了。

3、如果一个Consumer异常退出了,它处理的数据能够被另外的Consumer处理,这样数据在这种情况下就不会丢失了(注意是这种情况下)。

4、为了保证数据不被丢失,RabbitMQ支持消息确认机制,即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到,那么我们不能采用no-ack。而应该是在处理完数据后发送ack。

5、在处理数据后发送的ack,就是告诉RabbitMQ数据已经被接收,处理完成,RabbitMQ可以去安全的删除它了。

6、如果Consumer退出了但是没有发送ack,那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。

7、这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说,RabbitMQ给了Consumer足够长的时间来做数据处理。

    消息确认,对于spring-boot来说,就是一个开关,它就是spring.rabbitmq.listener.acknowledge-mode

    acknowledgeMode有三值:

    A、NONE = no acks will be sent (incompatible with channelTransacted=true).

          RabbitMQ calls this "autoack" because the broker assumes all messages are acked without any action from the consumer.

    B、MANUAL = the listener must acknowledge all messages by calling Channel.basicAck().

    C、AUTO = the container will acknowledge the message automatically, unless the MessageListener throws an exception.

Note that acknowledgeMode is complementary to channelTransacted - if the channel is transacted then the broker requires a commit notification in addition to the ack. This is the default mode. See also txSize.

    非常简单,在application.properties中增加spring.rabbitmq.listener.acknowledge-mode=AUTO

为了更好的演示异常,我们把生产者、消费者都做了sleep.代码如下:

sending

[html]  view plain copy
  1. package com.example;  
  2.   
  3. import java.util.Date;  
  4.   
  5. import org.slf4j.Logger;  
  6. import org.slf4j.LoggerFactory;  
  7. import org.springframework.amqp.core.AmqpTemplate;  
  8. import org.springframework.beans.factory.annotation.Autowired;  
  9. import org.springframework.stereotype.Component;  
  10.   
  11. @Component  
  12. public class HelloSender {  
  13.   
  14.     protected static Logger logger=LoggerFactory.getLogger(HelloSender.class);   
  15.       
  16.     @Autowired  
  17.     private AmqpTemplate rabbitTemplate;  
  18.   
  19.     public String send(String name) throws InterruptedException {  
  20.         String context = "hello "+name+" --" + new Date();  
  21.         String sendStr;  
  22.         for(int i=1;i<=100;i++){  
  23.             sendStr="第["+i+"]个 hello "+name+" --" + new Date();   
  24.             logger.debug("HelloSender: " + sendStr);  
  25.             this.rabbitTemplate.convertAndSend("hello", sendStr);  
  26.             Thread.sleep(1000);  
  27.         }  
  28.         return context;  
  29.     }  
  30. }  
Receiving

[html]  view plain copy
  1. package com.example;  
  2.   
  3. import org.slf4j.Logger;  
  4. import org.slf4j.LoggerFactory;  
  5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;  
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;  
  7. import org.springframework.stereotype.Component;  
  8.   
  9. @Component  
  10. @RabbitListener(queues = "hello")  
  11. public class HelloReceiver {  
  12.     protected static Logger logger = LoggerFactory.getLogger(HelloReceiver.class);  
  13.   
  14.     @RabbitHandler  
  15.     public void process(String hello) {  
  16.         logger.debug("HelloReceiver : " + hello);  
  17.         try {  
  18.             Thread.sleep(2000);  
  19.         } catch (InterruptedException e) {  
  20.             // TODO Auto-generated catch block  
  21.             e.printStackTrace();  
  22.         }  
  23.     }  
  24. }  

Receiving2

[html]  view plain copy
  1. package com.example;  
  2.   
  3. import org.slf4j.Logger;  
  4. import org.slf4j.LoggerFactory;  
  5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;  
  6. import org.springframework.amqp.rabbit.annotation.RabbitListener;  
  7. import org.springframework.stereotype.Component;  
  8.   
  9. @Component  
  10. @RabbitListener(queues = "hello")  
  11. public class HelloReceiver {  
  12.     protected static Logger logger = LoggerFactory.getLogger(HelloReceiver.class);  
  13.   
  14.     @RabbitHandler  
  15.     public void process(String hello) {  
  16.         logger.debug("HelloReceiver : " + hello);  
  17.         try {  
  18.             Thread.sleep(3000);  
  19.         } catch (InterruptedException e) {  
  20.             // TODO Auto-generated catch block  
  21.             e.printStackTrace();  
  22.         }  
  23.     }  
  24. }  
你会注意到两个消费者的sleep时间不一样,这是为了方便异常退出一个之后,查看另一个是否接收并处理。

四、消息持久化

在上一节中我们知道了即使Consumer异常退出,Message也不会丢失。但是如果RabbitMQ Server退出呢?软件都有bug,即使RabbitMQ Server是完美毫无bug的(当然这是不可能的,是软件就有bug,没有bug的那不叫软件),它还是有可能退出的:被其它软件影响,或者系统重启了,系统panic了。。。

为了保证在RabbitMQ退出或者crash了数据仍没有丢失,需要将queue和Message都要持久化。

queue持久化,就是在实例时调用具有参数durable的构造函数.

[java]  view plain copy
  1. package com.example;  
  2.   
  3. import org.springframework.amqp.core.Queue;  
  4. import org.springframework.context.annotation.Bean;  
  5. import org.springframework.context.annotation.Configuration;  
  6.   
  7. @Configuration  
  8. public class HelloRabbitConfig {  
  9.   
  10.     @Bean  
  11.     public Queue helloQueue() {  
  12.         return new Queue("hello",true);  
  13.     }  
  14. }  

五、单消费者,排他队列

个人理解: 这个是queue的重载构造方法,要使用单消费者,就得使用这个构造方法

消息只能一个消费者接收处理,其它消费者只能看着,这也是队列实例时调用具有参数exclusive 的构造函数。

    /**
     * Construct a new queue, given a name, durability, exclusive and auto-delete flags.
     * @param name the name of the queue.
     * @param durable true if we are declaring a durable queue (the queue will survive a server restart)
     * @param exclusive true if we are declaring an exclusive queue (the queue will only be used by the declarer's
     * connection)
     * @param autoDelete true if the server should delete the queue when it is no longer in use
     */
    public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete) {
        this(name, durable, exclusive, autoDelete, null);
    } 

六、参数


个人理解: 这个应该是配置文件里面的相关配置.

  • spring.rabbitmq.addresses指定client连接到的server的地址,多个以逗号分隔.

  • spring.rabbitmq.dynamic是否创建AmqpAdmin bean. 默认为: true)

  • spring.rabbitmq.host指定RabbitMQ host.默认为: localhost)

  • spring.rabbitmq.listener.acknowledge-mode指定Acknowledge的模式.

  • spring.rabbitmq.listener.auto-startup是否在启动时就启动mq,默认: true)

  • spring.rabbitmq.listener.concurrency指定最小的消费者数量.

  • spring.rabbitmq.listener.max-concurrency指定最大的消费者数量.

  • spring.rabbitmq.listener.prefetch指定一个请求能处理多少个消息,如果有事务的话,必须大于等于transaction数量.

  • spring.rabbitmq.listener.transaction-size指定一个事务处理的消息数量,最好是小于等于prefetch的数量.

  • spring.rabbitmq.password指定broker的密码.

  • spring.rabbitmq.port指定RabbitMQ 的端口,默认: 5672)

  • spring.rabbitmq.requested-heartbeat指定心跳超时,0为不指定.

  • spring.rabbitmq.ssl.enabled是否开始SSL,默认: false)

  • spring.rabbitmq.ssl.key-store指定持有SSL certificate的key store的路径

  • spring.rabbitmq.ssl.key-store-password指定访问key store的密码.

  • spring.rabbitmq.ssl.trust-store指定持有SSL certificates的Trust store.

  • spring.rabbitmq.ssl.trust-store-password指定访问trust store的密码.

  • spring.rabbitmq.username指定登陆broker的用户名.

  • spring.rabbitmq.virtual-host指定连接到broker的Virtual host.

这篇关于SpringBoot的RabbitMQ消息队列: 三、第二模式Work queues的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!


原文地址:
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.chinasem.cn/article/1095398

相关文章

Spring LDAP目录服务的使用示例

《SpringLDAP目录服务的使用示例》本文主要介绍了SpringLDAP目录服务的使用示例... 目录引言一、Spring LDAP基础二、LdapTemplate详解三、LDAP对象映射四、基本LDAP操作4.1 查询操作4.2 添加操作4.3 修改操作4.4 删除操作五、认证与授权六、高级特性与最佳

Spring Shell 命令行实现交互式Shell应用开发

《SpringShell命令行实现交互式Shell应用开发》本文主要介绍了SpringShell命令行实现交互式Shell应用开发,能够帮助开发者快速构建功能丰富的命令行应用程序,具有一定的参考价... 目录引言一、Spring Shell概述二、创建命令类三、命令参数处理四、命令分组与帮助系统五、自定义S

SpringKafka错误处理(重试机制与死信队列)

《SpringKafka错误处理(重试机制与死信队列)》SpringKafka提供了全面的错误处理机制,通过灵活的重试策略和死信队列处理,下面就来介绍一下,具有一定的参考价值,感兴趣的可以了解一下... 目录引言一、Spring Kafka错误处理基础二、配置重试机制三、死信队列实现四、特定异常的处理策略五

SpringSecurity JWT基于令牌的无状态认证实现

《SpringSecurityJWT基于令牌的无状态认证实现》SpringSecurity中实现基于JWT的无状态认证是一种常见的做法,本文就来介绍一下SpringSecurityJWT基于令牌的无... 目录引言一、JWT基本原理与结构二、Spring Security JWT依赖配置三、JWT令牌生成与

Java中Date、LocalDate、LocalDateTime、LocalTime、时间戳之间的相互转换代码

《Java中Date、LocalDate、LocalDateTime、LocalTime、时间戳之间的相互转换代码》:本文主要介绍Java中日期时间转换的多种方法,包括将Date转换为LocalD... 目录一、Date转LocalDateTime二、Date转LocalDate三、LocalDateTim

如何配置Spring Boot中的Jackson序列化

《如何配置SpringBoot中的Jackson序列化》在开发基于SpringBoot的应用程序时,Jackson是默认的JSON序列化和反序列化工具,本文将详细介绍如何在SpringBoot中配置... 目录配置Spring Boot中的Jackson序列化1. 为什么需要自定义Jackson配置?2.

Java中使用Hutool进行AES加密解密的方法举例

《Java中使用Hutool进行AES加密解密的方法举例》AES是一种对称加密,所谓对称加密就是加密与解密使用的秘钥是一个,下面:本文主要介绍Java中使用Hutool进行AES加密解密的相关资料... 目录前言一、Hutool简介与引入1.1 Hutool简介1.2 引入Hutool二、AES加密解密基础

Spring Boot项目部署命令java -jar的各种参数及作用详解

《SpringBoot项目部署命令java-jar的各种参数及作用详解》:本文主要介绍SpringBoot项目部署命令java-jar的各种参数及作用的相关资料,包括设置内存大小、垃圾回收... 目录前言一、基础命令结构二、常见的 Java 命令参数1. 设置内存大小2. 配置垃圾回收器3. 配置线程栈大小

SpringBoot实现微信小程序支付功能

《SpringBoot实现微信小程序支付功能》小程序支付功能已成为众多应用的核心需求之一,本文主要介绍了SpringBoot实现微信小程序支付功能,文中通过示例代码介绍的非常详细,对大家的学习或者工作... 目录一、引言二、准备工作(一)微信支付商户平台配置(二)Spring Boot项目搭建(三)配置文件

解决SpringBoot启动报错:Failed to load property source from location 'classpath:/application.yml'

《解决SpringBoot启动报错:Failedtoloadpropertysourcefromlocationclasspath:/application.yml问题》这篇文章主要介绍... 目录在启动SpringBoot项目时报如下错误原因可能是1.yml中语法错误2.yml文件格式是GBK总结在启动S