526互联

rabbitmq的消息持久化和消费者ack确认

发布时间 2023-07-11 15:07:23作者: 甜菜波波

消息持久化

队列持久化
在这里插入图片描述

交换机持久化

ExchangeBuilder.directExchange("normalExchange").build();
  • 1

在这里插入图片描述

消费者ack确认

multiple：批量
比如批量确认：

当multiple的值设置为true时，RabbitMQ将确认指定传输标签以及之前所有未被确认的消息。与单个确认相同，批量确认的作用域为每个通道。例如：通道Ch上有四个未被确认的消息，标签分别为5，6，7，8；当一个delivery_tag值为8并且multiple值为true的确认消息到达通道时，所有5到8的标签都会被确认。如果multiple值设置为false，标签为5，6，7的消息将不会被确认。

channel的各种ack

channel.basicAck(deliveryTag, false);

deliveryTag:该息的index
multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。

channel.basicNack(deliveryTag, false, true);

deliveryTag:该消息的index
multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息。
requeue：被拒绝的是否重新入队列

channel.basicReject(deliveryTag:, false);

deliveryTag:该消息的index
requeue：被拒绝的是否重新入队列
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15

消息正确消费了，使用channel.basicAck(deliveryTag, false)进行确认。
如果消息消费过程中发生异常，如果想把消息重新放回消息队列中给其他消费者或者下次消费，则使用 channel.basicNack(deliveryTag, false, true)，或者 channel.basicReject(deliveryTag:, true);如果消息异常，而且不想再把消息重新放入队列中，则使用 channel.basicNack(deliveryTag, false, false）, channel.basicReject(deliveryTag:, false);

maven依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
  • 1
  • 2
  • 3
  • 4

yaml配置

spring:
  application:
    name: ttl-queue
  rabbitmq:
    host: 110.40.181.73
    port: 35672
    username: root
    password: 10086
    virtual-host: /fchan
    connection-timeout: 15000
    # 发送确认
    publisher-confirm-type: none
    # 路由失败回调
    publisher-returns: true
    template:
      # 必须设置成true 消息路由失败通知监听者，而不是将消息丢弃
      mandatory: true
    listener:
      simple:
        # 每个channel通道每次从RabbitMQ获取的消息数量
        prefetch: 1
        #【 (若有重试)重试次数超过限制后】，是否将被拒绝的消息重新入队。在manual下没有ack确认，那么是不算拒绝的
        default-requeue-rejected: false
        # 每个队列启动的消费者数量,如果是工作队列模式那么就是每个监听者启动了50个消费者
        concurrency: 50
        # 每个队列最大的消费者数量
        max-concurrency: 50
        # 签收模式为手动签收-那么需要在代码中手动ACK
        acknowledge-mode: manual
  • 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

消费者监听

@RabbitListener(queues = "normalQueue2")
public void normalQueue3(Map<String,Object> map ,Message message, Channel channel) throws Exception {
    log.info("收到normalQueue3信息:{}",map);
    //当multiple的值设置为true时，RabbitMQ将确认指定传输标签以及之前所有未被确认的消息。
    // 与单个确认相同，批量确认的作用域为每个通道。
    // 例如：通道Ch上有四个未被确认的消息，标签分别为5，6，7，8；
    // 当一个delivery_tag值为8并且multiple值为true的确认消息到达通道时，所有5到8的标签都会被确认。
    // 如果multiple值设置为false，标签为5，6，7的消息将不会被确认。
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10