消费者从特定队列中接收消息的两种方式
1.持续订阅:将信道设置为接受模式,直到取消对队列的订阅为止。
(1)说明:消费者在接收(或拒绝)最近一条消息完成后,会自动接收下一条(可用的)消息,即消息已到达队列会自动接收
(2)使用AMQP的basic.consume命令
2.单条订阅:获得单条消息,然后取消订阅,使用AMQP的basic.get命令
队列分发消息给消费者
(1)当Rabbit队列中拥有多个消费者时,对列会将接收到的消息以循环的方式发送给消费者。
(2)每条消息只会发送给一个订阅的消费者。
(3)消费者接收到的每一条消息都必须进行确认。
(4)消费者确认后RabbitMQ才会吧消息从队列中删除
注意:
消费者对消息的确认和告知生产者消息已经被消费没有任何关系,确认只是消费者和队列之间的关系。
如果消费者受到一条消息后,没有确认就与Rabbit断开连接,RabbitMQ会认为该消息没有分发,会重新分配给另一个消费者;
如果消费者受到一条消息后,没有确认且未断开与Rabbit的连接,则RabbitMQ不会再给该消费者分发消息。可以应用于:消息内容处理耗时时,延迟消息确认,防止消息过载。
消费者拒绝收到某条消息的两种方式
(1)断开消费者和RabbitMQ之间的连接
(2)使用basic.reject命令
前提条件:RabbitMQ 2.0.0以上版本
将reject命令的queue设置为true,拒绝消息后RabbitMQ会将消息重新放回队列并分发给其他消费者
将reject命令的queue设置false,拒绝消息后RabbitMQ会将消息从队列中移除。可以应用于:检测到一条格式错误的消息谁也无法处理的时候。
创建队列
(1)生产者和消费者都能使用queue.declare命令
(2)当消费者在一条信道中订阅了一个队列,此时消费者无法在同一条信道上创建队列。解决办法:取消订阅,将信道设置为传输模式
(3)队列的名称,可以在创建时指定,若不指定,RabbitMQ会在queue.declare命令的返回中返回一个随机的队列名称,用于匿名队列
(4)exclusive:当设置为true时,队列为私有队列。用于限制一个队列只有一个消费者时。
(5)auto-delete:当最后一个消费者取消订阅时,队列就会被移除。用于临时队列。
注意:
当创建的队列已经存在时,若声明参数完全匹配,RabbitMQ不做任何操作直接返回成功(个人理解,有时候为了避免丢消息生产者和消费者都会尝试去声明队列,此为容错处理);若声明参数不匹配,则会返回错误
(6)检测队列是否存在:设置queue.declare的passive选线为true,若队列存在返回成功,失败返回错误
总结:队列是AMQP的基础模块,为消息提供了储存场所并均匀的分发给消费者,是负载均衡的最佳实现。
网友评论