mandatory参数

和路由键找到符合条件的队列#xff0c;rabbitmq调用Basic.Re…消息何去何从

mandatory和immediate是channel.basicPublish方法的两个参数都有消息传递过程中不可达目的地时将消息返回给生产者的功能。

mandatory参数

和路由键找到符合条件的队列rabbitmq调用Basic.Return命令将消息返回给生产者

生产者调用channel.addReturnListener添加ReturnListener监听器实现

false消息直接丢弃

告诉服务器至少将该消息路由到一个队列中否则将消息返回给生产者。

Rabbitmq3.0去掉了对immediate参数支持建议采用TTL和DLX方法替代

true如果交换器在将消息路由到队列时发现队列上并不存在任何消费者这条消息将不会存入队列中当与路由键匹配的所有队列都没有消费者时该消息会通过Basic.Return返回至生产者。

备份交换器AE

生产者发送消息不设置mandatory消息未被路由会丢失设置了需要添加ReturnListener。

如果不想编程复杂也不想消息丢失使用备份交换器。

channel.exchangeDeclare(“myAe”,“fanout”,true,false,null);channel.queueDeclare(“unroutedQueue”,true,false,false,null);channel.queueBind(“unroutedQueue”,“myAe”,“”);

rabbitmqctl

‘{“alternate-exchange”:“myAE”}’

特殊情况

如果设置了备份交换器不存在客户端和RabbitMQ服务端都不会有异常出现此时消息会丢失如果备份交换器没有绑定任何队列客户端和rabbitmq服务端都不会有异常出现此时消息会丢失如果备份交换器没有任何匹配的队列客户端和rabbitmq服务端都不会有异常出现此时消息会丢失如果备份交换器和mandatory参数一起使用那么mandatory参数无效

Live

通过队列属性设置队列中所有消息都有相同的过期时间一旦过期就从队列中抹去消息已经在队列头部只要定期从队列头部开始扫描即可

channel.queueDeclare方法中加入x-message-ttl参数实现单位ms

MapString,Object

channel.queueDeclare(queueName,durable,exclusive,autoDelete,args);通过Policy方式设置ttl

rabbitmqctl

通过对消息本身单独设置每条消息的ttl可以不同即使过期也不会马上抹去是否过期是在即将投递到消费者之前判定的

代码设置

设置AMQP.BasicProperties属性set属性deliveryMode持久化消息expirationttl时间

http

如果两个方法一起使用消息的ttl以两者之间较小的数值为准消息在队列中一旦超过设置的ttl时就会变成死信消费者将无法再收到该消息

不设置ttl表示此消息不会过期ttl0表示除非此时可以直接将消息投递到消费者否则立即丢弃。

设置队列TTL

channel.queueDeclare方法中的x-expires参数可以控制队列被自动删除前处于未使用状态的时间未使用队列上没有任何消费者队列也没有被重新声明并在过期时间段内也未调用过Basic.Get命令

MapString,

channel.queueDeclare(myqueue,false,false,false,args);死信队列

当消息在一个队列中变成死信后能被重新被发送到另一个交换器中这个交换器就是DLX死信交换器绑定DLX的队列就是死信队列

消息变成死信情况

当队列中存在死信时rabbitmq会自动将这个消息重新发布到设置的DLX上去进而被路由到另一个队列死信队列可以监听这个队列的消息进行相应处理

设置方法

channel.queueDeclare方法中设置x-dead-letter-exchange为队列添加DLX

延迟队列

延迟队列存储的对象是对应的延迟消息当消息被发送以后并不想让消费者立刻拿到消息而是等待特定时间后消费者才能拿到这个消息进行消费。

订单系统30min内未支付进行异常处理手机遥控设备指定时间工作

假设一个应用中需要将每条消息都设置为10秒延迟生产者通过exchange.normal交换器将发送的消息存储在queue.normal队列消费者订阅的是queue.dlx队列当消息从queue.normal整个队列中过期之后被存入queue.dlx队列消费者恰巧消费到了延迟10秒的这条消息。

优先级队列

前提如果在消费速度大于生成者的速度且broker中没有消息堆积的情况下对发送的消息设置优先级就没什么意义了。

RPC实现

客户端发送请求消息服务端回复响应的消息为了接收响应的消息需要在请求消息中发送一个回调队列

String

channel.queueDeclare().getQueue();

BasicProperties

BasicProperties.Builder().replyTo(callbackQueueName).build();

channel.basicPublish(,rpc_queue,props,message.getBytes());replyTo用来设置一个回调队列correlationId用来关联请求和其调用RPC之后的回复每一个请求设置一个唯一的correlationId

持久化

交换器持久化通过在声明队列是将durable参数置为true实现的。

如果不持久化rabbitmq服务重启后相关的交换器元数据会丢失消息不丢失只是不能将消息发送到这个交换器中了。

队列持久化通过在声明队列时将durable置为true如果不设置持久化rabbitmq重启后相关队列元数据会丢失此时数据也会丢失。

消息持久化通过将消息的投递模式BasicProperties中的diliveryMode属性设置为2即可实现消息的持久化设置了队列和消息的持久化rabbitmq服务重启后消息依旧存在。

生产者确认

确定消息到底有没有正确到达服务器。

可以通过事务机制和发送方确认机制

事务机制

channel.txSelect用于当前的信道设置成事务模式channel.txCommit用于提交事务channel.txRollback用于事务回滚

开启事务流程

客户端发送Tx.select将信道置为事务模式Broker回复Tx.Select-Ok,确认已将信道置为事务模式在发送完消息后客户端发送Tx.Commit提交事务Broker回复Tx.Commit-Ok确认提交事务如果发生异常在捕获异常后channel.txRollback()回滚

缺点会有性能损失

生产者将信道设置成confirm模式channel.confirmSelectrabbitmq同意Confirm.Select-Ok一旦信道进入confirm模式所有在该信道上发布的消息都会被指派一个唯一id一旦消息被投递到匹配的队列后rabbitmq会发送一个确认给生产者包含消息唯一id使得生产者知晓消息已经正确到达了目的地。

事务机制在一条消息发送后会使发送端阻塞等待rabbitmq回应后才发下一条消息而发送发确认机制最大好处是异步的。

生产者通过回调方法处理该确认消息。

如果rabbitmq因自身内部错误导致消息丢失会发送一条nack命令生产者应用程序同样可以在回调方法中处理nack命令。

publisher

批量confirm方法每发送一批消息后调用channel.waitForConfirms方法等待服务器的确认返回异步confirm方法提供一个回调方法服务端确认了一条或多条消息后客户端会回调这个方法进行处理。

消费端要点介绍

rabbitmq队列拥有多个消费者时队列收到的消息将以轮询的分发方式发送给消费者每条消息只会发送给订阅列表里的一个消费者。

问题如果某些空闲某些忙碌造成整体下降方法channel.basicQos方法允许限制信道上的消费者所能保持的最大未确认消息的数量。

如果达到上限就不会向这个消费者再发送任何消息知道消费者确认了某条消息后相应计数减1之后消费者可以继续接受消息。

消息顺序性

如果生产者使用了事务机制发送消息遇到异常进行了事务回滚需重新补偿发送如果是另一个线程实现则出现乱序。

如果生产者发送的消息设置了不同的超时时间并设置了死信队列顺序不一致。

设置了优先级也不是顺序的。

要保证消息的顺序性需要业务方使用rabbitmq之后进一步处理例如在消息体内添加全局有序标识实现。

弃用QueueingConsumer

内存溢出问题队列堆积较多的消息导致消费者客户端内存溢出假死不断堆积

会拖累同一个connection下的所有信道性能降低同步递归调用QueueingConsumer会产生死锁rabbitmq的自动连接恢复机制不支持Queueing

Consumer这种形式QueueingConsumer不是事件驱动的

消息传输保障

rabbitmq支持其中的最多一次和最少一次其中最少一次投递实现需要考虑

confirm机制以确保消息可以可靠地传输到rabbitmq中消息生产者需要配合使用mandatory参数或者备份交换器来确保消息能够从交换器路由到队列中进而能够保存下来而不会被丢弃消息和队列都需要进行持久化处理以确保rabbitmq服务器在遇到异常情况时不会造成消息丢失消费者在消费消息的同时需要将autoAck设置为false然后通过手动确认的方式去确认已经正确消费的消息以避免在消费端引起不必要的消息丢失。

参考《RabbitMQ实战指南》