当前位置：首页 > news >正文

RabbitMQ(二)队列与消息的持久化

news 来源：原创 2024/5/4 7:01:57

当有多个消费者同时收取消息，且每个消费者在接收消息的同时，还要做其它的事情，且会消耗很长的时间，在此过程中可能会出现一些意外，比如消息接收到一半的时候，一个消费者宕掉了，这时候就要使用消息接收确认机制，可以让其它的消费者再次执行刚才宕掉的消费者没有完成的事情。另外，在默认情况下，我们创建的消息队列以及存放在队列里面的消息，都是非持久化的，也就是说当RabbitMQ宕掉了或者是重启了，创建的消息队列以及消息都不会保存，为了解决这种情况，保证消息传输的可靠性，我们可以使用RabbitMQ提供的消息队列的持久化机制。

生产者：

1 import com.rabbitmq.client.ConnectionFactory;
2 import com.rabbitmq.client.Connection;
3 import com.rabbitmq.client.Channel;
4 import com.rabbitmq.client.MessageProperties;
5 public class ClientSend1 {
6      public static final String queue_name = " my_queue " ;
7      public static final boolean durable = true ; // 消息队列持久化
8      public static void main(String[] args)
9      throws java.io.IOException{
10         ConnectionFactory factory = new ConnectionFactory(); // 创建连接工厂
11         factory.setHost( " localhost " );
12         factory.setVirtualHost( " my_mq " );
13         factory.setUsername( " zhxia " );
14         factory.setPassword( " 123456 " );
15         Connection connection = factory.newConnection(); // 创建连接
16         Channel channel = connection.createChannel(); // 创建信道
17          channel.queueDeclare(queue_name, durable, false , false , null ); // 声明消息队列，且为可持久化的
18         String message = " Hello world " + Math.random();
19          // 将队列设置为持久化之后，还需要将消息也设为可持久化的，MessageProperties.PERSISTENT_TEXT_PLAIN
20          channel.basicPublish( "" , queue_name, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
21         System.out.println( " Send message: " + message);
22         channel.close();
23         connection.close();
24     }
25
26 }

说明：

行17 和行20 需要同时设置，也就是将队列设置为持久化之后，还需要将发送的消息也要设置为持久化才能保证队列和消息一直存在

消费者：

1 import com.rabbitmq.client.ConnectionFactory;
2 import com.rabbitmq.client.Connection;
3 import com.rabbitmq.client.Channel;
4 import com.rabbitmq.client.QueueingConsumer;
5 public class ClientReceive1 {
6      public static final String queue_name = " my_queue " ;
7      public static final boolean autoAck = false ;
8      public static final boolean durable = true ;
9      public static void main(String[] args)
10      throws java.io.IOException,java.lang.InterruptedException{
11         ConnectionFactory factory = new ConnectionFactory();
12         factory.setHost( " localhost " );
13         factory.setVirtualHost( " my_mq " );
14         factory.setUsername( " zhxia " );
15         factory.setPassword( " 123456 " );
16         Connection connection = factory.newConnection();
17         Channel channel = connection.createChannel();
18         channel.queueDeclare(queue_name, durable, false , false , null );
19         System.out.println( " Wait for message " );
20          channel.basicQos( 1 ); // 消息分发处理
21         QueueingConsumer consumer = new QueueingConsumer(channel);
22         channel.basicConsume(queue_name, autoAck, consumer);
23          while ( true ){
24             Thread.sleep( 500 );
25             QueueingConsumer.Delivery deliver = consumer.nextDelivery();
26             String message = new String(deliver.getBody());
27             System.out.println( " Message received: " + message);
28             channel.basicAck(deliver.getEnvelope().getDeliveryTag(), false );
29         }
30     }
31 }

说明：

行22：设置RabbitMQ调度分发消息的方式，也就是告诉RabbitMQ每次只给消费者处理一条消息，也就是等待消费者处理完并且已经对刚才处理的消息进行确认之后，才发送下一条消息，防止消费者太过于忙碌。如下图所示：