RabbitMQ之队列与消息持久化

偏执的太偏执、 2022-08-24 05:25 275阅读 0赞

# 队列持久化 #

在之前的例子中，我们所用的队列都是临时队列，当服务重启后之前创建的队列就都没有了。

队列的持久化是在定义队列时的**第二个参数**决定的(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后队列依然存在

![SouthEast][]

消息持久化的测试方法：

把消费者中确认接收消息的代码注释掉（前边提到过该操作），启动发送消息程序，重启RabbitMQ Server后消息依然可以接收到

[SouthEast]: /images/20220823/f518d7f446fb46f8a22da7f6ed9f20f7.png