RabbitMQ消息持久化处理

我们来看一下RabbitMQ的消息处理,我们先来看第一个知识点,关于RabbitMQ持久化的消息处理,消息的可靠性是RabbitMQ的一大特色,那么RabbitMQ是如何保证消息的可靠性的呢,就是靠他的消息持久化,那么什么是消息持久化处理呢,在这里我给大家做一个简单的介绍,比如我现在的消息提供者,不停的向我的消息消费者发送消息,在某一个时间段内,或者在某一个时间点上,突然我的消息消费者宕机了,然后经过抢修以后,这个服务又立即启动了,那么就在宕机和启动的时间轴上,这个时候的消息,时间段所发出的消息,那么对于消费者来讲,对于消息消费者来讲,他是否还能够接收得到这个消息了呢,答案肯定是可以接收的到的,最主要是依赖于Rabbit消息持久化的技术,接下来我们就通过编写这样一个案例,来展示消息持久化的一个体现

首先我们去创建一个Provider,创建一个Consumer,我们就拿direct交换器来作为案例交换器,你用fanout,topic都可以,我就那direct,我就先去做一个Provider,在这里我们给他起一个名字,在项目名当中,我们加一个durable,表示是一个持久的,rabbitmq-durable-direct-provider然后我们再去创建一个Consumerrabbitmq-durable-direct-consumer两个项目创建好以后呢,我们先去修改一下他的pom文件,在pom文件当中呢, 把artifactId和name,做一个更改,这样我们两个项目就创建好了,创建项目,我们先看我们拷贝过来的项目,在这里我们两个消息的接收者,一个是ErrorReceiver,一个是InfoReceiver,然后他们都是根据自己的路由器,决定从哪个队列里去取消息,然后我们去看消息的提供者,Sender,现在我们用的是error的路由key@Value("${mq.config.queue.error.routing.key}")
private String routingkey;我们看配置文件mq.config.queue.error.routing.key=log.error.routing.key那么按照这个的发现规则来看呢,我们应该是ErrorReceiver,接收到消息,然后我们再看一下测试代码,测试代码当中呢,我们这里有一个while死循环,然后里面休眠了一秒,一秒发送一条消息,这个时候应该会被谁接收呢,被我们的ErrorReceiver接收,我们发送是有他来发送的,代码的一个整体结构,现在我在这里做一个代码的修改,我首先让他休眠两秒,让他慢一点,这样看的清楚一点,然后我在外面设置一个int的变量,是0,我在这里flag++,加到我们的消息当中,什么意思呢,给我们的消息加一个编号,这样知道是第几条消息,然后我们再打开我们的管理页面59.110.158.145:15672大家还记得这个管理页面吧,然后我们看这个queues,现在这里是没有任何队列的,我们接下来先去启动消费者,消费者启动成功,我们再来启动消息的发送者,找到他的测试代码,我们来运行他,这个时候就两秒发送一次消息,两秒发送一次消息,这个时候ErrorReceiver已经接受消息了,两秒钟发送一次,两秒钟接收一次,然后我们再看管理页面,这个时候queues就直接有变化了,这个时候是不是出现了两个队列,一个好似log.error,一个是log.info,只不过我们现在发送消息是log.error,不是log.info,所以现在只有一个队列里有数据流,然后注意看他的Features,特征,现在特征是一个AD,鼠标往那里一放,auto-delete:true,那么auto-delete是什么含义呢,别着急,我们一会再来讲解他,给大家演示一下消息丢失的现象,怎么办呢,现在显示的消费者的线程,我把消息消费者关掉,你看我在哪里关,点,然后我们的消息在第51条停止了,宕机了,消费者宕机了,然后我的发送者这一侧,由于它是在一个死循环里,所以消息还是源源不断的在发送,我们这个时候再点queues,这里就已经动态更新了是不是又变成了no queues的动态了

现在没有队列了,你那个原来的log.error,log.info,队列都没有了,然后现在再看,我再把消费者启动,注意我们是在51条停的,运行走,这个时候我们就接收到消息了,第208条,这里这个队列就出来了,那么也就是说,从51到208之间的消息,就没有了,丢失了,这就是一个典型的消息丢失现象,那我们现在之所以会出现丢失消息的原因是什么呢,关键在这,看我们的Error Receiver,我们现在在设置这个Queue的时候,我们就是用了一个@Queue的一个注解,这里有一个autoDelete属性,然后我们给的是true,然后当时我们给的属性的时候,你就可以理解为一个临时队列,那么什么叫临时队列呢,如果我们在Queue里,autoDelete除了可以在@Queue的注解当中,在@Exchange这里也有一个autoDelete,也有这个属性,他里面赋的值也是true或者false,那么我们就把autoDelete这个属性设置一下,如果我们在@Queue注解上,加了autoDelete,这里是什么含义呢,当所有消费客户端,链接断开后,是否自动删除队列,如果设置成true,表示删除,如果设置成false,就表示不删除,看到了吗,我们现在只要所有的客户端,都不再连接这个队列了,这个队列就会被自动的删除,如果你设置true的话,如果设置成false呢,即便是所有的客户端都断开连接了,除了@Queue这一块可以设置,在@Exchange的注解里,也有autoDelete属性,如果我们要在这个注解里设置了,表示什么含义呢,当绑定队列都不再使用时,是否自动删除交换器,同样的道理,true表示删除,false表示不删除,其实你在哪个注解里设置,删队列还是删交换器,明白这个意思吧,所以@Queue作为我们讲解的重点,比如我们要解决丢失消息的问题,怎么办呢,我们刚才也看到了,其实之所以会产生消息丢失的原因就是,当我们的消费者服务一停掉,那么对于RabbitMQ来讲,会自动的把这个queue删除掉,所以我们再次启动的时候,在另一个Queue是没法拿到另一个消息的,所以我们要解决这个问题,很简单,当你的消费客户端,断开连接以后,RabbitMQ也不会删除这个Queue,那当我再次启动的时候,由于这个Queue还存在,那我们是不是还可以继续从这个Queue里去获取刚才我没有获取的消息,我们再来演示一下,现在我把这个true改成false@RabbitListener(bindings=@QueueBinding(value=@Queue(value="${mq.config.queue.error}",autoDelete="false"),exchange=@Exchange(value="${mq.config.exchange}",type=ExchangeTypes.DIRECT),key="${mq.config.queue.error.routing.key}")
)然后我们先去启动消费者,然后再去启动消息的提供者,我们看这个时候消息就来了,我们再去看管理界面,这个时候已经刷新了,然后注意看这个Features,由于我们只是在log.error的消息队列里,我们只改了这个的autoDelete,没有改log.info的,所以log.info还是AD状态的,就是当客户端,所有客户端一断开以后,而现在在log.error这一块,他的Feature里面,里面没有AD了,表示他是一个持久的队列,他并不会随着你客户端的关闭而删除,我们来看一下是不是这样的,我们现在还是把服务停掉,在第28条消息给关掉了,我们把它清空一下,然后我们的消息提供者还是在不断地发送消息,回过来,我们再去请求消费者,等一下我们先去看这个队列,看这个管理界面,是不是log.info没有了,但是log.error一直存在,而且还有数据向队列里进行输入,然后注意看这个ready,这个Ready是什么呢,其实就是对于未接收到的数据的一个显示,也就是说RabbitMQ,在队列里存放的消息,如果消息并没有被消费者所消费,那么他就会给这个消息加一个标记,表示当前这个消息是未被消费的,那我们现在看到的这个Ready里面,显示当前有多少条消息没有被消费,现在已经变成27条了,然后我们现在再去启动消费者,运行,观察他的控制台,看到了吗,他启动以后,我们从28条我们停了,从第29条开始读到消息,一直往后,读到现在,现在ready是不是没有了,现在已经没有被消费的消息,都被消费到了,所以我们解决了丢失消息的现象,只要是你的服务停了,只要你把autoDelete设置成false了,服务再启动的时候,由于队列,还存在,所以这些消息我们仍然可以接收到,我么之所以接收到的原因就是因为队列存在了,如果我们关闭以后,我们的消费者关闭以后,其实发送者发送的消息一直在那里存着呢,就是在RabbitMQ服务器的内存当中存放着,因为队列也在内存当中,队列里包含了这些数据,都在内存里存放着,但是我的RabbitMQ,整个服务都关掉了,服务都关掉了,那队列肯定也没了,消息肯定也没了,这就是我们采用autoDelete来对知识点的讲解,消息持久化的内容就到这

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.learn</groupId><artifactId>rabbitmq-durable-direct-consumer</artifactId><version>0.0.1-SNAPSHOT</version><packaging>jar</packaging><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>1.5.12.RELEASE</version><relativePath/> </parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version><thymeleaf.version>3.0.9.RELEASE</thymeleaf.version><thymeleaf-layout-dialect.version>2.2.2</thymeleaf-layout-dialect.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><!-- 这个插件,可以将应用打包成一个可执行的jar包 --><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

spring.application.name=rabbitmq-durable-direct-consumerspring.rabbitmq.host=59.110.158.145
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
mq.config.exchange=log.direct
mq.config.queue.info=log.info
mq.config.queue.info.routing.key=log.info.routing.key
mq.config.queue.error=log.error
mq.config.queue.error.routing.key=log.error.routing.key

package com.learn;import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;/*** 消息接收者* @author Administrator* @RabbitListener bindings:绑定队列* @QueueBinding  value:绑定队列的名称*                exchange:配置交换器* * @Queue value:配置队列名称*        autoDelete:是否是一个可删除的临时队列* * @Exchange value:为交换器起个名称*           type:指定具体的交换器类型*/
@Component
@RabbitListener(bindings=@QueueBinding(value=@Queue(value="${mq.config.queue.error}",autoDelete="false"),exchange=@Exchange(value="${mq.config.exchange}",type=ExchangeTypes.DIRECT),key="${mq.config.queue.error.routing.key}"))
public class ErrorReceiver {/*** 接收消息的方法。采用消息队列监听机制* @param msg*/@RabbitHandlerpublic void process(String msg){System.out.println("Error..........receiver: "+msg);}
}

package com.learn;import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;/*** 消息接收者* @author Administrator* @RabbitListener bindings:绑定队列* @QueueBinding  value:绑定队列的名称*                exchange:配置交换器* * @Queue value:配置队列名称*        autoDelete:是否是一个可删除的临时队列* * @Exchange value:为交换器起个名称*           type:指定具体的交换器类型*/
@Component
@RabbitListener(bindings=@QueueBinding(value=@Queue(value="${mq.config.queue.info}",autoDelete="true"),exchange=@Exchange(value="${mq.config.exchange}",type=ExchangeTypes.DIRECT),key="${mq.config.queue.info.routing.key}"))
public class InfoReceiver {/*** 接收消息的方法。采用消息队列监听机制* @param msg*/@RabbitHandlerpublic void process(String msg){System.out.println("Info........receiver: "+msg);}
}

package com.learn;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class RabbitDurableDirectConsumerApplication {public static void main(String[] args) {SpringApplication.run(RabbitDurableDirectConsumerApplication.class, args);}
}

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.learn</groupId><artifactId>rabbitmq-durable-direct-provider</artifactId><version>0.0.1-SNAPSHOT</version><packaging>jar</packaging><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>1.5.12.RELEASE</version><relativePath/> </parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version><thymeleaf.version>3.0.9.RELEASE</thymeleaf.version><thymeleaf-layout-dialect.version>2.2.2</thymeleaf-layout-dialect.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies><!-- 这个插件,可以将应用打包成一个可执行的jar包 --><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

spring.application.name=rabbitmq-durable-direct-providerspring.rabbitmq.host=59.110.158.145
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guestmq.config.exchange=log.direct
mq.config.queue.info.routing.key=log.info.routing.key
mq.config.queue.error.routing.key=log.error.routing.key
mq.config.queue.error=log.error

package com.learn;import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;/*** 消息发送者* @author Administrator**/
@Component
public class Sender {@Autowiredprivate AmqpTemplate rabbitAmqpTemplate;//exchange 交换器名称@Value("${mq.config.exchange}")private String exchange;//routingkey 路由键@Value("${mq.config.queue.error.routing.key}")private String routingkey;/** 发送消息的方法*/public void send(String msg){//向消息队列发送消息//参数一：交换器名称。//参数二：路由键//参数三：消息this.rabbitAmqpTemplate.convertAndSend(this.exchange, this.routingkey, msg);}
}

package com.learn;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class RabbitDurableDirectProviderApplication {public static void main(String[] args) {SpringApplication.run(RabbitDurableDirectProviderApplication.class, args);}
}

RabbitMQ消息持久化处理相关推荐

RabbitMQ 入门系列（10）— RabbitMQ 消息持久化、不丢失消息
消息要保持"持久化",即不丢失,必须要使得消息.交换器.队列,必须全部 "持久化". 1. 生产者怎么确认 RabbitMQ 已经收到了消息? # 打开通道的确 ...
RabbitMQ消息持久化
当RabbitMQ服务器挂了,它可能就丢失所有队列中的消息和任务.如果你想让RabbitMQ记住当前的状态和内容,就需要通过2件事来确保消息和任务不会丢失:同时将queue和messages标识为du ...
rabbitmq消息持久化,避免异常情况下,消息会丢失
2019独角兽企业重金招聘Python工程师标准>>> 1) 使用python包amqp from amqp.basic_message import Message from am ...
rabbitmq 持久化_RabbitMQ原理与相关操作(三)消息持久化
现在聊一下RabbitMQ消息持久化: 问题及方案描述 1.当有多个消费者同时收取消息,且每个消费者在接收消息的同时,还要处理其它的事情,且会消耗很长的时间.在此过程中可能会出现一些意外,比如消息接收 ...
rabbitmq 同步策略_RabbitMQ(三):消息持久化策略
一.前言在正常的服务器运行过程中,时常会面临服务器宕机重启的情况,那么我们的消息此时会如何呢?很不幸的事情就是,我们的消息可能会消失,这肯定不是我们希望见到的结果.所以我们希望AMQP服务器崩溃了也 ...
RabbitMQ之消息持久化
消息的可靠性是RabbitMQ的一大特色,那么RabbitMQ是如何保证消息可靠性的呢--消息持久化. 为了保证RabbitMQ在退出或者crash等异常情况下数据没有丢失,需要将queue,exch ...
RabbitMQ 队列消息持久化
参考链接: https://www.cnblogs.com/Keep-Ambition/p/8044752.html 假如消息队列test里面还有消息等待消费者(consumers)去接收,但是这个时 ...
rabbitmq消息队列，消息发送失败，消息持久化，消费者处理失败相关
转:https://blog.csdn.net/u014373554/article/details/92686063 项目是使用springboot项目开发的,前是代码实现,后面有分析发送消息失败. ...
RabbitMQ消息队列，发送消息失败、消息持久化、消费者失败处理方法和发送消息
项目是使用springboot项目开发的,前是代码实现,后面有分析发送消息失败.消息持久化.消费者失败处理方法和发送消息解决方法及手动确认的模式先引入pom.xml <!--rabbitmq- ...

RabbitMQ消息持久化处理

RabbitMQ消息持久化处理相关推荐

最新文章

热门文章