RabbitMQ实战教程(八) : 队列声明queueDeclare

忘是亡心i 2021-09-27 12:20 376阅读 0赞

> 原文出处 : [https://blog.csdn.net/vbirdbest/article/details/78670550][https_blog.csdn.net_vbirdbest_article_details_78670550]

## 简介 ##

本节主要讨论队列声明的各个参数

![这里写图片描述][SouthEast]

queueDeclare(String queue, 
                boolean durable, 
                boolean exclusive, 
                Map<String, Object> arguments);
     
      
       1
       2
       3
       4

*  queue: 队列名称
 *  durable： 是否持久化, 队列的声明默认是存放到内存中的，如果rabbitmq重启会丢失，如果想重启之后还存在就要使队列持久化，保存到Erlang自带的Mnesia数据库中，当rabbitmq重启之后会读取该数据库
 *  exclusive：是否排外的，有两个作用，一：当连接关闭时connection.close()该队列是否会自动删除；二：该队列是否是私有的private，如果不是排外的，可以使用两个消费者都访问同一个队列，没有任何问题，如果是排外的，会对当前队列加锁，其他通道channel是不能访问的，如果强制访问会报异常：com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: `#method<channel.close>(reply-code=405, reply-text=RESOURCE_LOCKED - cannot obtain exclusive access to locked queue 'queue_name' in vhost '/', class-id=50, method-id=20)`一般等于true的话用于一个队列只能有一个消费者来消费的场景
 *  autoDelete：是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除，可以通过RabbitMQ Management，查看某个队列的消费者数量，当consumers = 0时队列就会自动删除
 *  arguments：  
    队列中的消息什么时候会自动被删除？

*  Message TTL(x-message-ttl)：设置队列中的所有消息的生存周期(统一为整个队列的所有消息设置生命周期), 也可以在发布消息的时候单独为某个消息指定剩余生存时间,单位毫秒, 类似于redis中的ttl，生存时间到了，消息会被从队里中删除，注意是消息被删除，而不是队列被删除， 特性Features=TTL, 单独为某条消息设置过期时间AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties().builder().expiration(“6000”);  
    channel.basicPublish(EXCHANGE\_NAME, “”, properties.build(), message.getBytes(“UTF-8”));
 *  Auto Expire(x-expires): 当队列在指定的时间没有被访问(consume, basicGet, queueDeclare…)就会被删除,Features=Exp
 *  Max Length(x-max-length): 限定队列的消息的最大值长度，超过指定长度将会把最早的几条删除掉， 类似于mongodb中的固定集合，例如保存最新的100条消息, Feature=Lim
 *  Max Length Bytes(x-max-length-bytes): 限定队列最大占用的空间大小， 一般受限于内存、磁盘的大小, Features=Lim B
 *  Dead letter exchange(x-dead-letter-exchange)： 当队列消息长度大于最大长度、或者过期的等，将从队列中删除的消息推送到指定的交换机中去而不是丢弃掉,Features=DLX
 *  Dead letter routing key(x-dead-letter-routing-key)：将删除的消息推送到指定交换机的指定路由键的队列中去, Feature=DLK
 *  Maximum priority(x-max-priority)：优先级队列，声明队列时先定义最大优先级值(定义最大值一般不要太大)，在发布消息的时候指定该消息的优先级， 优先级更高（数值更大的）的消息先被消费,
 *  Lazy mode(x-queue-mode=lazy)： Lazy Queues: 先将消息保存到磁盘上，不放在内存中，当消费者开始消费的时候才加载到内存中
 *  Master locator(x-queue-master-locator)

## 注意 ##

关于队列的声明，如果使用同一套参数进行声明了，就不能再使用其他参数来声明，要么删除该队列重新删除，可以使用命令行删除也可以在RabbitMQ Management上删除，要么给队列重新起一个名字。

## 队列持久化 ##

重启RabbitMQ服务器(可以通过rabbitmqctl stop\_app关闭服务器，rabbitmqctl start\_app重启服务器)，可以登录RabbitMQ Management—> Queues中可以看到之前声明的队列还存在

boolean durable = true;
    channel.queueDeclare(QUEUE_NAME, durable, false, false, arguments);
     
      
       1
       2

## 消息持久化 ##

设置消息持久化必须先设置队列持久化，要不然队列不持久化，消息持久化，队列都不存在了，消息存在还有什么意义。消息持久化需要将交换机持久化、队列持久化、消息持久化，才能最终达到持久化的目的

方式一：设置deliveryMode=2

channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true);
    channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
    String message = "Hello RabbitMQ: ";
    // 设置消息持久化
    AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties().builder();
    properties.deliveryMode(2); // 设置消息是否持久化，1： 非持久化 2：持久化
    
    channel.basicPublish(EXCHANGE_NAME, "", properties.build(), message.getBytes("UTF-8"));
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10

方式二：设置BasicProperties为MessageProperties.PERSISTENT\_TEXT\_PLAIN

channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);
    channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
    String message = "Hello RabbitMQ: ";
    channel.basicPublish(EXCHANGE_NAME, "", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
     
      
       1
       2
       3
       4
       5

## Message TTL消息剩余生存时间 ##

统一设置队列中的所有消息的过期时间，例如设置10秒，10秒后这个队列的消息清零

方式一：为该队列的所有消息统一设置相同的声明周期

Map<String, Object> arguments = new HashMap<String, Object>();
    arguments.put("x-message-ttl", 10000);
    
    // 声明队列时指定队列中的消息过期时间
    channel.queueDeclare(QUEUE_NAME, false, false, false, arguments);
    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
     
      
       1
       2
       3
       4
       5
       6

方式二：单独为某条消息单独设置时间

// expiration: 设置单条消息的过期时间

channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
    
    for(int i = 1; i <= 5; i++) { 
       AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties()
           .builder().expiration( i * 1000 + "");
    
       channel.basicPublish(EXCHANGE_NAME, "", properties.build(), (message + i).getBytes("UTF-8"));
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9

## Auto Expire自动过期 ##

x-expires用于当多长时间没有消费者访问该队列的时候，该队列会自动删除，可以设置一个延迟时间，如仅启动一个生产者，10秒之后该队列会删除，或者启动一个生产者，再启动一个消费者，消费者运行结束后10秒，队列也会被删除

Map<String, Object> arguments = new HashMap<String, Object>();
    arguments.put("x-expires", 10000);
    
    channel.queueDeclare(QUEUE_NAME, false, false, false, arguments);
    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
     
      
       1
       2
       3
       4
       5

## Max Length最大长度 ##

x-max-length:用于指定队列的长度，如果不指定，可以认为是无限长，例如指定队列的长度是4，当超过4条消息，前面的消息将被删除，给后面的消息腾位

Map<String, Object> arguments = new HashMap<String, Object>();
    arguments.put("x-max-length", 4);
    
    channel.queueDeclare(QUEUE_NAME, false, false, false, arguments);
    for(int i = 1; i <= 5; i++) {
         channel.basicPublish(EXCHANGE_NAME, "", MessageProperties.PERSISTENT_TEXT_PLAIN, (message + i).getBytes("UTF-8"));
    }
     
      
       1
       2
       3
       4
       5
       6
       7

## Max Length Bytes代码片段 ##

x-max-length-bytes: 用于指定队列存储消息的占用空间大小，当达到最大值是会删除之前的数据腾出空间

Map<String, Object> arguments = new HashMap<String, Object>();
    rguments.put("x-max-length-bytes", 1024);
    
    channel.queueDeclare(QUEUE_NAME, false, false, false, arguments);
    channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
     
      
       1
       2
       3
       4
       5

## Maximum priority最大优先级 ##

x-max-priority: 设置消息的优先级，优先级值越大，越被提前消费。

正常情况下不适用优先级  
Hello RabbitMQ: 1  
Hello RabbitMQ: 2  
Hello RabbitMQ: 3  
Hello RabbitMQ: 4  
Hello RabbitMQ: 5

使用优先级顺序正好相反  
Hello RabbitMQ: 5  
Hello RabbitMQ: 4  
Hello RabbitMQ: 3  
Hello RabbitMQ: 2  
Hello RabbitMQ: 1

Map<String, Object> arguments = new HashMap<String, Object>();
    arguments.put("x-max-priority", 10);
    
    channel.queueDeclare(QUEUE_NAME, false, false, false, arguments);
    for(int i = 1; i <= 5; i++) {
        AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties()
            .builder().priority(i);
         channel.basicPublish(EXCHANGE_NAME, "", properties.build(), (message + i).getBytes("UTF-8"));
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9

![这里写图片描述][SouthEast 1]

## Dead letter exchange(死亡交换机) 和 Dead letter routing key(死亡路由键) ##

当队列中的消息过期，或者达到最大长度而被删除，或者达到最大空间时而被删除时，可以将这些被删除的信息推送到其他交换机中，让其他消费者订阅这些被删除的消息，处理这些消息

public void testBasicPublish() throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(AMQP.PROTOCOL.PORT);
        factory.setUsername("mengday");
        factory.setPassword("mengday");
    
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
    
        // 声明一个接收被删除的消息的交换机和队列
        String EXCHANGE_DEAD_NAME = "exchange.dead";
        String QUEUE_DEAD_NAME = "queue_dead";
        channel.exchangeDeclare(EXCHANGE_DEAD_NAME, BuiltinExchangeType.DIRECT);
        channel.queueDeclare(QUEUE_DEAD_NAME, false, false, false, null);
        channel.queueBind(QUEUE_DEAD_NAME, EXCHANGE_DEAD_NAME, "routingkey.dead");
    
        String EXCHANGE_NAME = "exchange.fanout";
        String QUEUE_NAME = "queue_name";
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
    
        Map<String, Object> arguments = new HashMap<String, Object>();
        arguments.put("x-message-ttl", 15000);
        arguments.put("x-max-length", 4);
        arguments.put("x-max-length-bytes", 1024);
        arguments.put("x-expires", 30000);
    
        arguments.put("x-dead-letter-exchange", "exchange.dead");
        arguments.put("x-dead-letter-routing-key", "routingkey.dead");
        channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
        String message = "Hello RabbitMQ: ";
        for(int i = 1; i <= 5; i++) {
            channel.basicPublish(EXCHANGE_NAME, "", null, (message + i).getBytes("UTF-8"));
        }
    
    
        channel.close();
        connection.close();
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41

刚开始由于队列长度是4，总共发送5条消息，所以最早进入队列的消息1将被删除掉，被推送到“死亡队列”中，所以看到普通队列的消息为4条，死亡队列的消息为1条，是消息1  
![这里写图片描述][SouthEast 2]

随着时间的流逝，普通队列中的消息都该过期了，所以消息2、3、4、5都被推送到死亡队列，所以死亡队列消息是5条，普通队列的消息条数为0  
![这里写图片描述][SouthEast 3]

再随着时间的流逝，普通队列过了指定时间没有被消费者访问，这个队列自动被删除了，所以看不到普通队列了，只有死亡队列  
![这里写图片描述][SouthEast 4]

查看死亡队列的消息可以得知，消息一死亡的原因是maxlen达到了最大长度，消息2、3、4、5都是因为生存时间到了导致死亡的  
![这里写图片描述][SouthEast 5]  
![这里写图片描述][SouthEast 6]

## 一个比较杂的综合示例 ##

关于消费者就不用代码来获取消息了，直接在RabbitMQ Management点击某个队列的名字，然后Get Message(s) 即可获取

该示例使用很多参数配置，可能实际使用不会像这样用，因为这样好像不太配套。

public class Producer {
        @Test
        public void testBasicPublish() throws IOException, TimeoutException, InterruptedException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(AMQP.PROTOCOL.PORT);
            factory.setUsername("mengday");
            factory.setPassword("mengday");
    
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            // 声明一个接收被删除的消息的交换机和队列
            String EXCHANGE_DEAD_NAME = "exchange.dead";
            String QUEUE_DEAD_NAME = "queue_dead";
            channel.exchangeDeclare(EXCHANGE_DEAD_NAME, BuiltinExchangeType.DIRECT);
            channel.queueDeclare(QUEUE_DEAD_NAME, false, false, false, null);
            channel.queueBind(QUEUE_DEAD_NAME, EXCHANGE_DEAD_NAME, "routingkey.dead");
    
            String EXCHANGE_NAME = "exchange.fanout";
            String QUEUE_NAME = "queue_name";
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
    
    
            Map<String, Object> arguments = new HashMap<String, Object>();
            // 统一设置队列中的所有消息的过期时间
            arguments.put("x-message-ttl", 30000);
            // 设置超过多少毫秒没有消费者来访问队列，就删除队列的时间
            arguments.put("x-expires", 20000);
            // 设置队列的最新的N条消息，如果超过N条，前面的消息将从队列中移除掉
            arguments.put("x-max-length", 4);
            // 设置队列的内容的最大空间，超过该阈值就删除之前的消息
            arguments.put("x-max-length-bytes", 1024);
            // 将删除的消息推送到指定的交换机，一般x-dead-letter-exchange和x-dead-letter-routing-key需要同时设置
            arguments.put("x-dead-letter-exchange", "exchange.dead");
            // 将删除的消息推送到指定的交换机对应的路由键
            arguments.put("x-dead-letter-routing-key", "routingkey.dead");
            // 设置消息的优先级，优先级大的优先被消费
            arguments.put("x-max-priority", 10);
            channel.queueDeclare(QUEUE_NAME, false, false, false, arguments);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
    
            String message = "Hello RabbitMQ: ";
            for(int i = 1; i <= 5; i++) {
                // expiration: 设置单条消息的过期时间
                AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties().builder().priority(i).expiration( i * 1000 + "");
                channel.basicPublish(EXCHANGE_NAME, "", properties.build(), (message + i).getBytes("UTF-8"));
            }
    
    
            channel.close();
            connection.close();
        }
    }
    
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42
       43
       44
       45
       46
       47
       48
       49
       50
       51
       52
       53
       54
       55
       56

## 运行效果 ##

![这里写图片描述][SouthEast 7]

![这里写图片描述][SouthEast 8]

![这里写图片描述][SouthEast 9]

![这里写图片描述][SouthEast 10]

![这里写图片描述][SouthEast 11]

![这里写图片描述][SouthEast 12]

疑惑: 单独使用arguments.put(“x-max-length”, 4); arguments.put(“x-dead-letter-exchange”, “exchange.dead”);arguments.put(“x-dead-letter-routing-key”, “routingkey.dead”);发现消息1先会触发maxlen条件，而被推送到queue\_dead队列中，由此可以得出，当达到最大长度时，先删除的是先被添加到队列的消息。但是如果很多条件一块同时使用可能现象不太好解释，如上例如，实际结果是消息5因为maxlen而被推送到死亡队列中，消息1、2、3、4都是由于expired过期导致的，难道不是消息1由于maxlen被推送到希望队列，而2、3、4、5是由于过期导致的吗？还有上面代码如果将x-max-len该为3，在死亡队列中获取消息的先后顺序是4、5、3、2、1不是优先级高的先被消费吗，为啥不是5、4、3、2、1 难道是条件用的太多了，都乱了？？？ 明白的同学请留言，谢谢！

经过实际测试，参数单独用或者和其他参数合理搭配使用都没问题，如果是像上例一块都用，大杂烩，搞不懂结果。

// 检查队列是否存在，不存在抛异常  
channel.queueDeclarePassive(“queue\_name”);

[https_blog.csdn.net_vbirdbest_article_details_78670550]: https://blog.csdn.net/vbirdbest/article/details/78670550
[SouthEast]: /images/20210923/c045e2ec79254db3a2f123e1f457e69e.png
[SouthEast 1]: /images/20210923/133a6de08b67426cb98eaf3560422c45.png
[SouthEast 2]: /images/20210923/38af801b185c463c9265517f8fd1d534.png
[SouthEast 3]: /images/20210923/d8178558f72c4a99b0be57567a972c48.png
[SouthEast 4]: /images/20210923/e01844bdf70e4343ba5a39ba0365c6ff.png
[SouthEast 5]: /images/20210923/bc8a75dccfbe43919542946ff3de3aaa.png
[SouthEast 6]: /images/20210923/22947da5719b42f191141e000ab0677f.png
[SouthEast 7]: /images/20210923/fe2b8637c69645d8806677920ba2d4e3.png
[SouthEast 8]: /images/20210923/b2276111a0c54af496e9150bb36ecc83.png
[SouthEast 9]: /images/20210923/4b7573660b4042cc87af3f15a7a33a1c.png
[SouthEast 10]: /images/20210923/ac7decbefdef4f38a43250eb1fb4e4a2.png
[SouthEast 11]: /images/20210923/8056562514834bc292bc93545bec7120.png
[SouthEast 12]: /images/20210923/522de995324c4d8ea8b433f4ad9d7dbf.png