RabbitMq高级特性-1

一、消息的可靠传递
①发送方的确认

在使用RabbitMQ的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ为我们提供了两种方式用来控制消息的投递可靠性模式。

1. confirm确认模式
2. return退回模式

rabbitmq整个消息投递的路径为:

producer—>rabbitmq broker—>exchange—>queue—>consumer

• 消息从producer 到 exchange则会返回一个confirmCallback 。
• 消息从exchange–>queue投递失败则会返回一个returnCallback。

1.1 确认模式的开启
配置文件application.yml中

spring:
  rabbitmq:
 	publisher-confirm-type: correlated
1
2
3

参数：
ack:是否成功发送到exchange
cause：发送失败的原因
correlationData：相关的配置信息

 rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("comfirmCallback被执行了");
                if(ack){
                    System.out.println("确认发送成功");
                }else {
                    System.out.println("发送失败，原因:"+cause);
                }
            }
        });
1
2
3
4
5
6
7
8
9
10
11

1.2退回模式
配置文件application.yml

spring:
  rabbitmq:
      publisher-returns: true
1
2
3

参数：
message：消息对象
replycode:错误码
replyText:错误信息
exchange:交换机
routingkey:路由键

rabbitTemplate.setMandatory(true); // 设置交换器处理失败消息的模式
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingkey) {
                System.out.println(message);
            }
        });
1
2
3
4
5
6
7

②接收方的确认ack

• 自动确认:acknowledge=“none”
• 手动确认: acknowledge=“manual”
• 根据异常情况确认: acknowledge=“auto”

2.1手动确认ack
basicAck(deliveryTag,multiple)
basicNack(deliveryTag,mutiple,requeue）

---

deliveryTag:
消息的编号,它是一个64位的长整型值。

mutiple:
如果multiple参数设置为false，则表示拒绝编号为deliveryTag的这一条消息，如果multiple参数设置为true则表示拒绝deliveryTag编号之前所有未被确认的消息。

requeue：
设置为true，则Rabbit MQ会重新将这条消息存入队列，以便可以发送给下一个订阅的消费者；如果requeue参数设置为false，则Rabbit MQ立即会把消息从队列中移除，而不会发送给新的消费者。

public void receiveMsg1(Message message, Channel channel) throws InterruptedException, IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            System.out.println(messageConverter.fromMessage(message));
            //处理业务
            System.out.println("处理业务.....");
            System.out.println("同意签收");
            channel.basicAck(deliveryTag, true);

        } catch (Exception e) {
            System.out.println("拒绝签收");
            channel.basicNack(deliveryTag, true, true);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

二、消费端的限流

1、设置ack为手动
2、设置prefech,一次从mq中拉取10条消息消费

spring:
  rabbitmq:
    listener:
      simple:
        prefetch: 10
1
2
3
4
5

三、TTL（队列过期时间）

 @Bean
    public Queue ttlQueue(){
        return QueueBuilder.durable().ttl(10000).build();
    }
1
2
3
4

通过参数MessagePostProcessor

 rabbitTemplate.convertAndSend("hzx.directExchange", "yellow", msg, new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setExpiration("5000");
                    return  message;
                }
            });
1
2
3
4
5
6
7