RabbitMQ常见问题

Posted 2022-02-05 假装Java大神

RabbitMQ常见问题

• RabbitMQ持久化问题
• RabbitMQ消息可靠生产
• RabbitMQ消费者异常死循环问题
• RabbitMQ如何保障消息可靠消费
• RabbitMQ死信队列
• • 消费失败
  • 消息过期TTL
  • 队列存储界限

RabbitMQ持久化问题

首先我们需要知道两个熟悉:
1.durable: 是否开启持久化，true是持久化队列(默认)，false非持久化队列
2.autoDelete: 是否为临时队列，true是临时队列当服务停止运行的时候会将队列进行删除，false是非临时队列(默认)

如果配置了durable:true(队列持久化) autoDelete:true(临时队列)，那么服务关闭的时候队列也会消失，会造成消息丢失

如果配置了durable:false(队列非持久化) autoDelete:false(非临时队列) 那么当rabbitMQ服务重启过后队列也会消失，同样会造成消息丢失

所以为了避免消息丢失需要将durable熟悉设置为true,autoDelete熟悉设置为false(当然这是默认配置好了)

---

RabbitMQ消息可靠生产

解决方案流程图如下

修改配置文件

# 开启消息发送确认机制
spring:
	rabbitmq:
		publisher-confirms: true

实现代码如下

	@Autowired
    private RabbitTemplate rabbitTemplate;
    // 交换机名称
    @Value("$log.direct")
    private String exchange;


	/**
	* 消息生产
	*/
	public void logMessage(String routingKey)
        // 设置回调确认对象
        rabbitTemplate.setConfirmCallback(confirmCallback);
        // 消息内容
        String msg = "路由模式,时间：" + new Date();
        // 相关数据
        CorrelationData correlationData = new CorrelationData();
        // 设置id
        correlationData.setId("1001");
        // 发送消息
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, correlationData);
    

    // 定义回调确认对象
    private RabbitTemplate.ConfirmCallback confirmCallback =
                new RabbitTemplate.ConfirmCallback() 
        // 消息发送完毕后，回调此确认方法
        @Override
        public void confirm(CorrelationData correlationData, boolean ack,
                            String cause) 
            // CorrelationData: 相关数据
            // 生产消息的时候设置好的id 在这边可以进行获取出来
            correlationData.getId();
            // ack: 是否确认收到(true已确认收到，false未确认收到)
            // case: 失败原因
            System.out.println("ack: " + ack);
            System.out.println("cause = " + cause);
            // 如果ack为true,代表MQ已经收到消息。
            if (ack)
                System.out.println("消息已投递成功！");
            else
            
                System.out.println("消息已投递失败: " + correlationData.getId());
                // 失败的消息业务处理代码
                // ...
            
        
    ;

---

RabbitMQ消费者异常死循环问题

当消费者获取数据进行消费的时候，必然会处理相关的业务。由于消息数据的不正确必然会导致报错与消息消费失败，就会一直重新消费。
本人亲自踩过的坑，当消息一直在重复消费结果导致到最后面一台消费者所部属的服务器直接挂了😂

那么如何避免消息重复消费呢？
我们只需要修改配置文件，设置最大重试次数即可

# 配置rabbitmq
spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true # 解决消息死循环问题-启用重试
          max-attempts: 3 # 最大重试3次(默认)

---

RabbitMQ如何保障消息可靠消费

需要开启手动消息确认ACK: 如果在处理消息的过程中，消费者在消费消息的时候服务器、网络、出现故障挂掉了，那可能这条正在处理的消息就没有完成，数据就会丢失。为了确保消息不会丢失，RabbitMQ支持消息确认ACK。

需要在配置文件开启手动确认ACK

# 配置rabbitmq
spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual # 开启手动ack消息确认

实现代码

@Component
public class ErrorConsumer 

    /**
     * 消息监听方法
     * bindings: 配置队列 通过路由key绑定到 交换机
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "$direct.error.queue"), // 队列
            key = "info","error", // 路由key
            exchange = @Exchange(name = "$log.direct",
                                 type = ExchangeTypes.DIRECT))) // 交换机
    public void handlerMessage(String msg, Channel channel,
                               Message message)
        try 
            System.out.println("================");
            // 制造异常
            int i = 10 / 0;
            System.out.println("error--->接受到的消息是：" + msg);
            // 手动ack确认
            //参数1：deliveryTag:消息唯一传输ID
            //参数2：multiple：true: 手动批量处理，false: 手动单条处理
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        catch (Exception ex)
        
    

---

RabbitMQ死信队列

消费失败

接着上面的问题,如果这个时候消费者确实失败了又不能一直堵塞消息通道那么该怎么处理？

• 死信队列 + 消费预警 + 记录到redis数据库

1.修改yml配置

# 配置rabbitmq
spring:
  rabbitmq:
    listener:
      simple:
        default-requeue-rejected: false # 设置为false，会重发消息到死信队列

2.定义死信队列消费者

@Component
public class DeadLetterConsumer 

    /** 消息监听 */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "dlx-queue"),
            exchange = @Exchange(value = "dlx.exchange",
                        type = ExchangeTypes.TOPIC),
            key = "#")
    )
    public void handlerMessage(String message)
        // todo 存到redis / mysql 都可以
        System.out.println("死信队列接收到的消息：" + message);
    

3.业务消息消费之

/**
     * 消息监听方法
     * bindings: 配置队列 通过路由key绑定到 交换机
     */
    @RabbitListener(bindings = @QueueBinding(
          value = @Queue(name = "$direct.error.queue", arguments = 
                  @Argument(name="x-dead-letter-exchange", value = "dlx.exchange"),// 死信队列交换机
                  @Argument(name="x-dead-letter-routing-key", value = "xxx")), // 死信队列路由队列
            key = "info","error", // 路由key
            exchange = @Exchange(name = "$log.direct",
                                 type = ExchangeTypes.DIRECT))) // 交换机
    public void handlerMessage(String msg, Channel channel,
                               Message message)
        try 
            System.out.println("================");
            // 制造异常
            int i = 10 / 0;
            System.out.println("error--->接受到的消息是：" + msg);
            // 手动ack确认
            // 参数1：deliveryTag:消息唯一传输ID
            // 参数2：multiple：true: 手动批量处理，false: 手动单条处理
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        catch (Exception ex)
            // 如果真得出现了异常，我们采用消息重投
            // 获取redelivered，判断是否为重投: false没有重投，true重投
            Boolean redelivered = message.getMessageProperties().getRedelivered();
            System.out.println("redelivered = " + redelivered);
            try 
                // 判断是否为重新消费
                if (redelivered)  // 重新消费
                    /**
                     * 拒绝确认，从队列中删除该消息，防止队列阻塞(消息堆积)
                     * boolean requeue: false不重新入队列(丢弃消息)
                     */
                    channel.basicReject(message.getMessageProperties()
                                        .getDeliveryTag(), false);
                    System.out.println("消息已重新存入死信队列了。。。");
                 else  // 第一次消费

                    /**
                     * 消息重投，重新把消息放回队列中
                     * boolean multiple: 单条或批量
                     * boolean requeue: true重回队列
                     */
                    channel.basicNack(message.getMessageProperties()
                            .getDeliveryTag(), false, true);

                
            catch (Exception e)
                e.printStackTrace();
            
        
    

---

消息过期TTL

可以设置消息超过多少毫秒还没消费就进入死信队列

队列存储界限

设置队列中存储消息的最大数量