rabbitmq----------队列

布满荆棘的人生 2022-12-22 09:18 105阅读 0赞

### 1、临时队列 ###

临时队列对应的是没有持久化的队列，也就是如果 RabbitMQ 服务器重启，那么这些队列就不会存在，所以我们称之为临时队列。

### 2、自动删除队列 ###

自动删除队列和普通队列在使用上没有什么区别，唯一的区别是，当消费者断开连接时，队列将会被删除。自动删除队列允许的消费者没有限制，  
也就是说当这个队列上最后一个消费者断开连接才会执行删除。  
自动删除队列只需要在声明队列时，设置属性 auto-delete 标识为 true 即可。系统声明的随机队列，缺省就是自动删除的。

//源码
 public com.rabbitmq.client.AMQP.Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete/*为true时，为自动删除*/, Map<String, Object> arguments) throws IOException {
 com.rabbitmq.client.AMQP.Queue.DeclareOk ok = this.delegate.queueDeclare(queue, durable, exclusive, autoDelete, arguments);
 RecordedQueue q = (new RecordedQueue(this, ok.getQueue())).durable(durable).exclusive(exclusive).autoDelete(autoDelete).arguments(arguments);
 if (queue.equals("")) {
 q.serverNamed(true);
 }
 
 this.recordQueue((com.rabbitmq.client.AMQP.Queue.DeclareOk)ok, q);
 return ok;
 }

3、单消费者队列

普通队列允许的消费者没有限制，多个消费者绑定到多个队列时，RabbitMQ 会采用轮询进行投递。如果需要消费者独占队列，在队列创建的时候，  
设定属性 exclusive 为 true。

![20201119101839299.png][]

4、自动过期队列

指队列在超过一定时间没使用，会被从 RabbitMQ 中被删除。  
什么是没使用？1.一定时间内没有 Get 操作发生。2.没有 Consumer 连接在队列上。  
特别的：就算一直有消息进入队列，也不算队列在被使用。  
通过声明队列时，设定 x-expires 参数即可，单位毫秒。  
比如这个队列会在当消费者断开连接时 10 秒后过期。

![20201119102112540.png][]

5、永久队列

持久化队列和非持久化队列的区别是，持久化队列会被保存在磁盘中，固定并持久的存储，当 Rabbit 服务重启后，该队列会保持原来的状态在 RabbitMQ  
中被管理，而非持久化队列不会被保存在磁盘中，Rabbit 服务重启后队列就会消失。  
非持久化比持久化的优势就是，由于非持久化不需要保存在磁盘中，所以使用速度就比持久化队列快。即是非持久化的性能要高于持久化。而持久  
化的优点就是会一直存在，不会随服务的重启或服务器的宕机而消失。  
在声明队列时，将属性 durable 设置为“false”，则该队列为非持久化队列，设置成“true”时，该队列就为持久化队列

![20201119104259191.png][]

6、队列级别消息过期

就是为每个队列设置消息的超时时间。只要给队列设置 x-message-ttl 参数，就设定了该队列所有消息的存活时间，时间单位是毫秒。如果声明队列  
时指定了死信交换器，则过期消息会成为死信消息。

![20201119104514633.png][]

7、队列保留参数列表

<table style="width:500pt;"> 
 <tbody> 
 <tr> 
 <td style="border-color:#FFFFFF;text-align:center;vertical-align:top;width:213pt;"> 参数名 </td> 
 <td style="border-color:#FFFFFF;text-align:center;vertical-align:top;width:287pt;"> 目的 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:213pt;"> x-dead-letter-exchange </td> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:287pt;"> 死信交换器 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:213pt;"> x-dead-letter-routing-key </td> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:287pt;"> 死信消息的可选路由键 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:213pt;"> x-expires </td> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:287pt;"> 队列在指定毫秒数后被删除 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:213pt;"> x-ha-policy </td> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:287pt;"> 创建HA队列 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:213pt;"> x-ha-nodes </td> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:287pt;"> HA队列的分布节点 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:213pt;"> x-max-length </td> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:287pt;"> 队列的最大消息数 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:213pt;"> x-message-ttl </td> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:287pt;"> 毫秒为单位的消息过期时间，队列级别 </td> 
 </tr> 
 <tr> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:213pt;"> x-max-priority </td> 
 <td style="border-color:#FFFFFF;vertical-align:top;width:287pt;"> 最大优先值为255的队列优先排序功能 </td> 
 </tr> 
 </tbody> 
</table>

8、消息的属性

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

9、消息存活时间

当队列消息的 TTL 和消息 TTL 都被设置，时间短的 TTL 设置生效。  
如果将一个过期消息发送给 RabbitMQ，该消息不会路由到任何队列，而是直接丢弃。  
为消息设置 TTL 有一个问题：RabbitMQ 只对处于队头的消息判断是否过期（即不会扫描队列），所以，很可能队列中已存在死消息，但是队列并不  
知情。这会影响队列统计数据的正确性，妨碍队列及时释放资源。

10、消息的持久化

默认情况下，队列和交换器在服务器重启后都会消失，消息当然也是。将队列和交换器的 durable 属性设为 true，缺省为 false，但是消息要持久化还  
不够，还需要将消息在发布前，将投递模式设置为 2。

生产者代码

//TODO 创建持久化交换器 durable=true
 channel.exchangeDeclare(EXCHANGE_NAME,"direct",true);
 
 /*日志消息级别，作为路由键使用*/
 String[] routekeys = {"king","mark","james"};
 for(int i=0;i<3;i++){
 String routekey = routekeys[i%3];
 String msg = "Hellol,RabbitMq"+(i+1);
 //TODO 发布持久化的消息(delivery-mode=2)
 channel.basicPublish(EXCHANGE_NAME,routekey,
 MessageProperties.PERSISTENT_TEXT_PLAIN,
 msg.getBytes());
 }

消息要持久化，必须要有持久化的队列、交换器和投递模式都为 2。

消费者代码

//TODO 声明一个持久化队列(durable=true)
            String queueName = "msgdurable";
            channel.queueDeclare(queueName,true,false,
                    false,null);

10、Request-Response 模式

我们前面的学习模式中都是一方负责发送消息而另外一方负责处理。而我们实际中的很多应用相当于一种一应一答的过程，需要双方都能给对方发  
送消息。于是请求-应答的这种通信方式也很重要。它也应用的很普遍。

生产者代码示例：

package cn.enjoyedu.setmsg;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.UUID;
    import java.util.concurrent.TimeoutException;
    
    /**
    
     *类说明：消息的属性的控制
     */
    public class ReplyToProducer {
    
        public final static String EXCHANGE_NAME = "replyto";
    
        public static void main(String[] args)
                throws IOException, TimeoutException {
            /* 创建连接,连接到RabbitMQ*/
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");
            Connection connection = connectionFactory.newConnection();
    
            /*创建信道*/
            Channel channel = connection.createChannel();
            /*创建持久化交换器*/
            channel.exchangeDeclare(EXCHANGE_NAME,"direct",false);
    
            //TODO 响应QueueName ，消费者将会把要返回的信息发送到该Queue
            String responseQueue = channel.queueDeclare().getQueue();
            //TODO 消息的唯一id
            String msgId = UUID.randomUUID().toString();
            //TODO 设置消息中的应答属性,建造者模式
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .replyTo(responseQueue)
                    .messageId(msgId)
                    .build();
    
            /*声明了一个消费者*/
            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("Received["+envelope.getRoutingKey()
                            +"]"+message);
                }
            };
            //TODO 消费者应答队列上的消息
            channel.basicConsume(responseQueue,true,consumer);
    
            String msg = "Hello,RabbitMq";
            //TODO 发送消息时，把响应相关属性设置进去
            channel.basicPublish(EXCHANGE_NAME,"error",
                    properties,
                    msg.getBytes());
            System.out.println("Sent error:"+msg);
        }
    
    }

消费者代码示例：

package cn.enjoyedu.setmsg;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
    
     *类说明：消息的属性的控制
     */
    public class ReplyToConsumer {
    
        public static void main(String[] argv)
                throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
    
            // 打开连接和创建频道，与发送端一样
            Connection connection = factory.newConnection();
            final Channel channel = connection.createChannel();
            /*创建交换器*/
            channel.exchangeDeclare(ReplyToProducer.EXCHANGE_NAME,
                    "direct",false);
    
            /*声明一个队列*/
            String queueName = "replyto";
            channel.queueDeclare(queueName,false,false,
                    false,null);
    
            /*绑定，将队列和交换器通过路由键进行绑定*/
            String routekey = "error";/*表示只关注error级别的日志消息*/
            channel.queueBind(queueName,ReplyToProducer.EXCHANGE_NAME,routekey);
    
            System.out.println("waiting for message........");
    
            /*声明了一个消费者*/
            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("Received["+envelope.getRoutingKey()
                            +"]"+message);
                    //TODO 从消息中拿到相关属性（确定要应答的消息ID,）
                    AMQP.BasicProperties respProp
                            = new AMQP.BasicProperties.Builder()
                            .replyTo(properties.getReplyTo())
                            .correlationId(properties.getMessageId())
                            .build();
                    //TODO 消息消费时，同时需要生作为生产者生产消息（以OK为标识）
                    channel.basicPublish("", respProp.getReplyTo() ,
                            respProp ,
                            ("OK,"+message).getBytes("UTF-8"));
               }
            };
            /*消费者正式开始在指定队列上消费消息*/
            channel.basicConsume(queueName,true,consumer);
    
    
        }
    
    }

[20201119101839299.png]: /images/20221120/4ee4da031c61431ea0b6a47a899b9500.png
[20201119102112540.png]: /images/20221120/8e2c881e31444ff088e5715b82cdc341.png
[20201119104259191.png]: /images/20221120/fa4ccedadfe541c6a6e800d8f9b14b1c.png
[20201119104514633.png]: /images/20221120/af6b0b8e9b1f4ea4a384b1fe3bfe6130.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221120/30eecd7c2c0747ae8400f92b3c80c4d8.png