文档章节

轻松搞定RabbitMQ(三)——消息应答与消息持久化

白志华
 白志华
发布于 2015/10/18 10:56
字数 1264
阅读 80
收藏 3

       这个官网的第二个例子中的消息应答和消息持久化部分。我把它摘出来作为单独的一块儿来分享。


Message acknowledgment(消息应答)

       执行一个任务可能需要花费几秒钟,你可能会担心如果一个消费者在执行任务过程中挂掉了。基于现在的代码,一旦RabbitMQ将消息分发给了消费者,就会从内存中删除。在这种情况下,如果杀死正在执行任务的消费者,会丢失正在处理的消息,也会丢失已经分发给这个消费者但尚未处理的消息。

       但是,我们不想丢失任何任务,如果有一个消费者挂掉了,那么我们应该将分发给它的任务交付给另一个消费者去处理。

       为了确保消息不会丢失,RabbitMQ支持消息应答。消费者发送一个消息应答,告诉RabbitMQ这个消息已经接收并且处理完毕了。RabbitMQ可以删除它了。

       如果一个消费者挂掉却没有发送应答,RabbitMQ会理解为这个消息没有处理完全,然后交给另一个消费者去重新处理。这样,你就可以确认即使消费者偶尔挂掉也不会不丢失任何消息了。

       没有任何消息超时限制;只有当消费者挂掉时,RabbitMQ才会重新投递。即使处理一条消息会花费很长的时间。

       消息应答是默认打开的。我们明确地把它们关掉了(autoAck=true)。现在将应答打开,一旦我们完成任务,消费者会自动发送消息应答。

boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);

       修改一下Worker.java

channel.basicQos(1);//保证一次只分发一个
		// 创建队列消费者
		final Consumer consumer = new DefaultConsumer(channel) {
			  @Override
			  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
			    String message = new String(body, "UTF-8");

			    System.out.println(" [x] Received '" + message + "'");
			    System.out.println(" [x] Proccessing... at " +new Date().toLocaleString());
			    try {
			    	for (char ch: message.toCharArray()) {
				        if (ch == '.') {
				        	Thread.sleep(1000);
				        }
				    }
				} catch (InterruptedException e) {
				} finally {
			      System.out.println(" [x] Done! at " +new Date().toLocaleString());
			      channel.basicAck(envelope.getDeliveryTag(), false);
			    }
			  }
			};
       我们还是运行1个生产者,2个消费者,在消息处理过程中,人为让一个消费者挂掉,然后会看到剩下的任务都会被另外的消费者执行。

       运行结果如下:


       如果你关闭了自动消息应答,手动也未设置应答,这是一个很简单错误,但是后果却是极其严重的。消息在分发出去以后,得不到回应,所以不会在内存中删除,结果RabbitMQ会越来越占用内存,最终的结果,你懂得。。。


Message durability(消息持久化)

       我们已经了解了如何确保即使消费者死亡,任务也不会丢失。但是如果RabbitMQ服务器停止,我们的任务仍将失去!
       当RabbitMQ退出或者崩溃,将会丢失队列和消息。除非你不要队列和消息。两件事儿必须保证消息不被丢失:我们必须把“队列”和“消息”设为持久化。 
    

boolean durable = true;
channel.queueDeclare("hello", durable, false, false, null);
       尽管这行代码是正确的,但他不会在我们当前的设置中起作用。因为我们已经定义了一个名叫hello的未持久化的队列。RabbitMQ不允许使用不同的参数设定重新定义已经存在的队列,并且会向尝试如此做的程序返回一个错误。一个快速的解决方案——就是声明一个不同名字的队列,比如task_queue。

       (当然,我们也可以登录到RabbitMQ的服务管理页面,RabbitMQ默认的端口是5672,管理页面默认端口是15672,页面地址为:http://localhost:15672,使用是用户名和密码登录。RabbitMQ的默认密码和用户名都是guest。点开“queue”那栏,可以看到队列列表,点击“hello”杜列,会展开队列的详细信息。把页面拉到最后,有一项“Delete / purge”,点开,点击“Delete”按钮,就可以把队列删除掉了。 然后再运行代码的时候,就会创建一个支持持久化的hello队列。)

       上述的代码需要在生产者和消费者都要作出同样的修改。

       在这一点上我们确信task_queue的队列不会丢失,即使RabbitMQ服务重启。现在我们需要将消息标记为持久性的——通过设置 MessageProperties(实现BasicProperties)为PERSISTENT_TEXT_PLAIN。

       现在你可以启动RabbitMQ服务器,执行一次生产者NewTask的程序,然后关闭服务器,再重新启动服务器,运行消费者Work做下实验。可以发现消费者依旧可以读出消息来。说明在RabbitMQ服务器关闭后,消息和队列信息都已经做了持久化。再次启动后,会重新加载到服务器中,消费者运行后,就可以正常的从队列中获取消息了。

本文转载自:http://blog.csdn.net/xiaoxian8023/article/details/48710653

共有 人打赏支持
白志华
粉丝 29
博文 265
码字总数 57524
作品 0
长沙
程序员
RabbitMQ实战2.消息轮询、响应、持久化

继上篇 RabbitMQ实战1.消息代理 消息轮询分配 如果生产者投递的消息需要运行相当长的时间,且有多个消费者在处理消息,那么RabbitMQ是怎么分配消息的? 新建 new_task.py 新建 worker.py 在三...

章鱼喵_
08/09
0
0
你不知道的RabbitMQ集群架构全解

你不知道的RabbitMQ集群架构全解 RabbitMQ系列文章 RabbitMQ在Ubuntu上的环境搭建 深入了解RabbitMQ工作原理及简单使用 RabbitMQ交换器Exchange介绍与实践 RabbitMQ事务和Confirm发送方消息确...

王磊的博客
07/25
0
0
RabbitMQ 入门【精+转】

rabbitmq可以用一本书取讲,这里只是介绍一些使用过程中,常用到的基本的知识点。 官方文档覆盖的内容,非常全面:http://www.rabbitmq.com/documentation.html 。 1. 介绍 RabbitMQ,即消息...

sunsky303
05/01
0
0
Rabbitmq-server 分析

一、 发送端: 接收端: 验证: 在多个shell窗口中运行consumer,然后运行procedure,发现每个consumer 轮流接收相同队列中的消息。 二、 发送端: 接收端: (1) rabbitmq循环调度,将消息循...

meteor_hy
05/31
0
0
PHP消息队列rabbitmq——windows

一.安装拓展 因为RabbitMQ是由erlang语言实现的,所以先要安装erlang环境 erlang 下载安装 http://www.erlang.org/download.html rabbitmq 下载安装 https://www.rabbitmq.com/install-windo...

小良下山化了个缘
09/07
0
0

没有更多内容

加载失败,请刷新页面

加载更多

OSChina 周四乱弹 —— 毒蛇当辣条

Osc乱弹歌单(2018)请戳(这里) 【今日歌曲】 @ 达尔文:分享花澤香菜/前野智昭/小野大輔/井上喜久子的单曲《ミッション! 健?康?第?イチ》 《ミッション! 健?康?第?イチ》- 花澤香菜/前野智...

小小编辑
48分钟前
4
1
java -jar运行内存设置

java -Xms64m #JVM启动时的初始堆大小 -Xmx128m #最大堆大小 -Xmn64m #年轻代的大小,其余的空间是老年代 -XX:MaxMetaspaceSize=128m # -XX:CompressedClassSpaceSize=6...

李玉长
58分钟前
1
0
Spring | 手把手教你SSM最优雅的整合方式

HEY 本节主要内容为:基于Spring从0到1搭建一个web工程,适合初学者,Java初级开发者。欢迎与我交流。 MODULE 新建一个Maven工程。 不论你是什么工具,选这个就可以了,然后next,直至finis...

冯文议
今天
1
0
RxJS的另外四种实现方式(四)——性能最高的库(续)

接上一篇RxJS的另外四种实现方式(三)——性能最高的库 上一篇文章我展示了这个最高性能库的实现方法。下面我介绍一下这个性能提升的秘密。 首先,为了弄清楚Most库究竟为何如此快,我必须借...

一个灰
今天
1
0
麒麟AI首席科学家现世

8月31日,华为发布了新一代顶级人工智能手机芯片麒麟980,成为全球首款7nm工艺手机芯片,AI方面也实现飞跃,支持人脸识别、物体识别、物体检测、图像分割、智能翻译等。 虽然如今人人都在热议...

问题终结者
昨天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部