RabbitMQ之队列与消息持久化
2014-06-27 22:12
330 查看
队列持久化
在之前的例子中,我们所用的队列都是临时队列,当服务重启后之前创建的队列就都没有了。队列的持久化是在定义队列时的第二个参数决定的(false为队列不用持久化)
channel.queueDeclare(queueName, false, false,false,null);如果持久化标志设置为true,则代表是一个持久的队列,那么在服务重启后,也会存在。因为服务会把持久化的queue存放在硬盘上,当服务重启的时候,会重新申明之前被持久化的queue。队列是可以被持久化,但是里面的消息是否为持久化那还要看消息的持久化设置。也就是说,如果重启之前那个queue里面还有没有发出去的消息的话,重启之后那队列里面是不是还存在原来的消息,这个就要取决于发送者在发送消息时对消息的设置了。
消息持久化
如果要在重启后保持消息的持久化必须设置消息是持久化的标志channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
示例代码
发送端:import java.io.IOException; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.MessageProperties; /** * 消息持久化 * @author menghh * */ public class Send04 { public static void main(String[] args) throws IOException { ConnectionFactory factory = new ConnectionFactory(); //RabbitMQ-Server安装在本机,所以直接用127.0.0.1 factory.setHost("127.0.0.1"); //创建一个连接 Connection conn = factory.newConnection(); //创建一个通信通道 Channel channel = conn.createChannel(); //定义Queue名称 String queueName = "queue01"; //为Channel定义queue的属性,queueName为queue名称 channel.queueDeclare(queueName, true, false,false,null); String msg = "Hello World!"; //发送消息 /** * 测试条件:1、在消息队列持久化的前提下2、接收消息方设置接收方式为手动接收,并不对接收消息进行确认 * 不采用消息持久化,重启RabbitMQ服务后,消息队列存在,消息接收不到 * 采用消息持久化,重启RabbitMQ服务后,消息队列存在,消息依然可以接收到消息,说明消息被持久化 */ channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes()); // channel.basicPublish("", queueName, null, msg.getBytes()); System.out.println("send message["+msg+"] to "+queueName+"success!"); //关闭通道 channel.close(); //关闭连接 conn.close(); } }
接收端(跟之前代码一样)
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.QueueingConsumer; import com.rabbitmq.client.QueueingConsumer.Delivery; /** * 消息持久化 * @author menghh * */ public class Recv04 { public static void main(String[] args) throws Exception{ ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); Connection conn = factory.newConnection(); Channel channel = conn.createChannel(); String queueName = "queue01"; channel.queueDeclare(queueName, true, false, false, null); //以上部分和sender一样 //配置好获取消息得方式 QueueingConsumer consumer = new QueueingConsumer(channel); channel.basicConsume(queueName, false,consumer); //循环获取消息 while(true){ //获取消息,如果没有消息,这一步将会一直阻塞 Delivery delivery = consumer.nextDelivery(); String msg = new String(delivery.getBody()); //确认消息,已经收到 channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false); System.out.println("received message["+msg+"] from "+queueName); } } }
测试结果:
运行程序后,队列存在,重启RabbitMQ Server后队列依然存在消息持久化的测试方法:
把消费者中确认接收消息的代码注释掉(前边提到过该操作),启动发送消息程序,重启RabbitMQ
Server后消息依然可以接收到
相关文章推荐
- RabbitMQ之消息持久化(队列持久化、消息持久化)
- RabbitMQ(二)队列与消息的持久化
- Java使用Rabbitmq惊喜队列queue和消息内容的本地持久化核心方法。(内容存储在硬盘)
- RabbitMQ队列与消息的持久化
- RabbitMQ系列三 (深入消息队列) 持久化
- RabbitMQ之队列与消息持久化
- RabbitMQ(二)队列与消息的持久化
- RabbitMQ之消息持久化(队列持久化、消息持久化)
- RabbitMQ3.7.2入门到进阶之工作队列消息应答ack与消息持久化durable
- RabbitMQ消息队列(一): Detailed Introduction 详细介绍
- RabbitMQ消息队列(五):Routing 消息路由
- ActiveMQ发消息、收消息、持久化,查询队列剩余消息数、出队数的实现
- RabbitMQ消息队列(一): Detailed Introduction 详细介绍
- RabbitMQ消息队列(七):适用于云计算集群的远程调用(RPC)
- RabbitMQ与SpringMVC集成并实现发送消息和接收消息(持久化)方案二
- RabbitMQ 用SpringBoot处理消息队列
- (十二)RabbitMQ消息队列-性能测试
- 关于消息队列的使用----ActiveMQ,RabbitMQ,ZeroMQ,Kafka,MetaMQ,RocketMQ
- C#中使用RabbitMQ收发队列消息
- 在C#中使用消息队列RabbitMQ