声明队列使用的queueDeclare()方法包含一系列入参,这些入参定义了队列的属性。需要注意的是,一旦队列创建后,其属性就不能改变了。若声明队列使用的入参与队列的属性不符,将会报错。
运行第2章的发送方代码,再将queueDeclare()中的入参中durable参数修改为true后运行,将会有如下报错:
Caused by: com.rabbitmq.client.ShutdownSignalException:
channel error; protocol method: #method<channel.close>
(reply-code=406, reply-text=PRECONDITION_FAILED -
inequivalent arg 'durable' for queue 'Queue_Java' in vhost '/':
received 'true' but current is 'false', class-id=50, method-id=10)
...
关键的报错信息为“inequivalent arg ‘durable’ for queue ‘Queue_Java’ in vhost ‘/’: received ‘true’ but current is ‘false’”,可见队列当前持久化(durable)的属性为false,可收到了true,导致报错。
以下逐一说明声明队列时各入参含义。
3.1 持久化
RabbitMQ的队列和消息存放于内存中,在出现任何异常情况,导致RabbitMQ异常中止,内存中的数据将会丢失。若要在RabbitMQ从异常恢复后仍然保存有异常中止前的消息,就需将消息持久化,即将消息同时保存于磁盘中。RabbitMQ可将消息保存到Erlang自带的Mnesia数据库中,当RabbitMQ重启之后会读取该数据库。
声明队列的方法中,“durable”参数即指定队列是否是持久化的。使消息持久化,需要队列和消息都是持久化的,并且通常交换机也应该是持久化的。RabbitMQ的默认交换机“(AMQP default)”是持久化的,对于与其绑定的队列,将队列声明为持久化的队列,并发送持久化的消息,即可将消息持久化。
// 声明持久化队列,durable = true
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 发送持久化消息
byte[] message = "This is Java's message".getBytes();
AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties().builder();
properties.deliveryMode(2);
channel.basicPublish(DEFAULT_EXCHANGE, QUEUE_NAME, properties.build(), message);
发送持久化消息后,重启RabbitMQ:
[[email protected] sbin]# ./rabbitmqctl stop_app
[[email protected] sbin]# ./rabbitmqctl start_app
登录Web管理界面,可以看到,队列中依然保存着重启前的消息。队列列表中,特性(Features)列的“D”即表示该队列是持久化的(Durable)。
可以验证,若队列不是持久化的,或发送的消息未设置持久化的属性,在RabbitMQ重启后,消息都会丢失。
3.2 排他
对于排他队列,只有创建它的连接有权访问,连接断开后,排他队列将自动删除。这种队列适用于一个队列仅对应一个客户端收发消息的场景。在声明队列时,将exclusive参数设置为true即可声明一个排他队列。
可通过以下程序验证排他队列的性质。
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
// 设置地址、端口、用户名、密码...
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明排他队列,exclusive = true
channel.queueDeclare(QUEUE_NAME, true, true, false, null);
// 等待控制台输入,以使连接保持
new Scanner(System.in).nextLine();
channel.close();
connection.close();
}
在程序运行至等待控制台输入时,进入Web管理界面,可以看到,队列列表中出现了被创建的排他队列,Features列的“Excl”即表明该队列是排他的。
保持等待控制台输入的状态下,启动另一个连接,以同样的参数声明该队列,会出现以下异常:
Caused by: 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_Java'
in vhost '/', class-id=50, method-id=10)
...
关键的提示信息为“cannot obtain exclusive access to locked queue”,无法获取加锁队列的访问权限,即该队列对于其他连接是无权访问的。
在控制台输入回车,使程序继续运行。连接关闭后,查看队列列表,可以看到,被创建的排他队列已被删除。
3.3 自动删除
若队列的autoDelete(自动删除)属性开启,当队列的最后一个消费者断开时,该队列会被自动删除。
声明一个队列,autoDelete参数设置为true,进入Web管理界面,可以看到,队列列表中出现了被创建的自动删除队列,Features列的“AD”即表明该队列是自动删除的。
点击队列名进入队列的管理界面,在“Consumers”下拉框中能看到连接该队列的所有消费者。
启动一个消费者程序连接该队列,连接后断开,可以看到,该队列在消费者断开后被删除了。
3.4 其他参数
在声明队列方法的入参中,存在一个称为“arguments”的参数,该参数以键值对的形式,可为队列设置多种属性。以“arguments”入参声明的队列,同样须以相同的属性和值为参数进行连接。可为队列设置以下参数。
x-message-ttl:队列中消息的生存周期,单位毫秒,设置该属性后,队列对每条消息自接收开始起计时,超过设置的时间值后,自动删除该消息;
x-expires:队列闲置时间,单位毫秒,若队列在指定的时间内没有被访问,则会自动删除;
x-max-length:队列的最大长度,即队列存放消息的最大数目,若接收消息后队列的消息数超过指定的队列最大长度,则将最早的消息移出队列;
x-max-length-bytes:队列最大占用空间,单位字节,若接收消息后队列的占用空间超过指定的队列最大占用空间,则将最早的消息移出队列;
x-dead-letter-exchange、x-dead-letter-routing-key:设置这两个参数后,当队列超过指定长度或占用空间大小时,对于移出的较早的消息,将其以指定的路由键发送至指定的交换机,而不直接删除,x-dead-letter-exchange指定交换机名称,x-dead-letter-routing-key指定路由键;
x-max-priority:优先级队列,指定队列的最大优先级,可在发布消息的时候指定消息的优先级,优先级更高(数值更大的)的消息先被消费,使用见后续消息章节;
x-queue-mode:队列模式,分为默认(default)和延迟(lazy)两种,当该值设置为“lazy”时,消息将先保存到磁盘上、不放在内存中,当消费者开始消费的时候才加载到内存中。
设置参数的示例如下。
Map<String, Object> arguments = new HashMap<String, Object>();
arguments.put("x-message-ttl", 10000); // 消息生存时间,10s
arguments.put("x-expires", 20000); // 队列闲置时间,20s
arguments.put("x-max-length", 100); // 队列最大长度,100条消息
arguments.put("x-max-length-bytes", 1024); // 队列最大占用空间,1024字节
// 移出的消息发送至“amq.direct”交换机,路由键为“dead”
arguments.put("x-dead-letter-exchange", "amq.direct");
arguments.put("x-dead-letter-routing-key", "dead");
arguments.put("x-max-priority", 10); // 最大优先级为10
arguments.put("x-queue-mode", "lazy"); // 延迟模式
channel.queueDeclare(QUEUE_NAME, true, false, false, arguments);