Skip to content

Latest commit

 

History

History
74 lines (55 loc) · 5.25 KB

如何解决消息队列的延时和过期失效，消息积压问题.md

File metadata and controls

74 lines (55 loc) · 5.25 KB

[toc]

如何解决消息队列的延时和过期失效，消息积压问题

  • 本质原因是消费端出问题，不消费了。
  • 或者消费的速度极其慢。
  • 消息堆积导致消息队列集群的磁盘快写满了，积压时间长，对于RabbitMQ设置了消息过期时间丢失怎么办？

大量消息在mq里积压了几个小时还没解决

  • 场景：几千万条数据在 MQ 里积压了七八个小时，从下午 4 点多，积压到了晚上 11 点多。
  • 修复了consumer问题，按照正常消费速度，需要几个小时。
    • 一个消费者一秒是 1000 条，一秒 3 个消费者是 3000 条，一分钟就是 18 万条。所以如果你积压了几百万到上千万的数据，即使消费者恢复了，也需要大概 1 小时的时间才能恢复过来。
  • 临时紧急扩容，具体操作步骤和思路如下：
    • 先修复 consumer 的问题，确保其恢复消费速度，然后将现有 consumer 都停掉
    • 新建一个 topic，partition 是原来的 10 倍，临时建立好原先 10 倍的 queue 数量
    • 然后写一个临时的分发数据的 consumer 程序，这个程序部署上去消费积压的数据，消费之后不做耗时的处理，直接均匀轮询写入临时建立好的 10 倍数量的 queue
    • 接着临时征用 10 倍的机器来部署 consumer，每一批 consumer 消费一个临时 queue 的数据。这种做法相当于是临时将 queue 资源和 consumer 资源扩大 10 倍，以正常的 10 倍速度来消费数据
    • 等快速消费完积压数据之后，得恢复原先部署的架构，重新用原先的 consumer 机器来消费消息

mq中的消息过期失效了

  • RabbtiMQ 是可以设置过期时间的，也就是 TTL。如果消息在 queue 中积压超过一定的时间就会被 RabbitMQ 给清理掉，这个数据就没了。大量数据积压在mq里，会导致大量的数据直接丢失。
  • 方案：批量重导
    • 写临时程序，把丢失的数据一点一点查出来，重新放入mq中。
    • 假设 1 万个订单积压在 mq 里面，没有处理，其中 1000 个订单都丢了，你只能手动写程序把那 1000 个订单给查出来，手动发到 mq 里去再补一次。

mq都快写满了

  • 临时写程序，接入来消费，消费一个丢弃一个，都不要了，快速消费掉所有的消息。
  • 然后到晚上再补数据。

RocketMQ官方针对消息积压问题，提供了解决方案：

1. 提高消费并行度

  • 绝大部分消息消费行为都属于 IO 密集型，即可能是操作数据库，或者调用 RPC，这类消费行为的消费速度在于后端数据库或者外系统的吞吐量，通过增加消费并行度，可以提高总的消费吞吐量，但是并行度增加到一定程度，反而会下降。所以，应用必须要设置合理的并行度。 如下有几种修改消费并行度的方法：
  • 同一个 ConsumerGroup 下，通过增加 Consumer 实例数量来提高并行度（需要注意的是超过订阅队列数的 Consumer 实例无效）。可以通过加机器，或者在已有机器启动多个进程的方式。 提高单个 Consumer 的消费并行线程，通过修改参数 consumeThreadMin、consumeThreadMax 实现。

2. 批量方式消费

  • 某些业务流程如果支持批量方式消费，则可以很大程度上提高消费吞吐量，例如订单扣款类应用，一次处理一个订单耗时 1 s，一次处理 10 个订单可能也只耗时 2 s，这样即可大幅度提高消费的吞吐量，通过设置 consumer 的 consumeMessageBatchMaxSize 返个参数，默认是 1，即一次只消费一条消息，例如设置为 N，那么每次消费的消息数小于等于 N。

3. 跳过非重要消息

  • 发生消息堆积时，如果消费速度一直追不上发送速度，如果业务对数据要求不高的话，可以选择丢弃不重要的消息。例如，当某个队列的消息数堆积到 100000 条以上，则尝试丢弃部分或全部消息，这样就可以快速追上发送消息的速度。示例代码如下：
public ConsumeConcurrentlyStatus consumeMessage(
            List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
    long offset = msgs.get(0).getQueueOffset();
    String maxOffset =
            msgs.get(0).getProperty(Message.PROPERTY_MAX_OFFSET);
    long diff = Long.parseLong(maxOffset) - offset;
    if (diff > 100000) {
        // TODO 消息堆积情况的特殊处理
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
    // TODO 正常消费过程
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

4. 优化每条消息消费过程

  • 举例如下，某条消息的消费过程如下：

    • 根据消息从 DB 查询【数据 1】
    • 根据消息从 DB 查询【数据 2】
    • 复杂的业务计算
    • 向 DB 插入【数据 3】
    • 向 DB 插入【数据 4】

  • 这条消息的消费过程中有 4 次与 DB 的 交互，如果按照每次 5ms 计算，那么总共耗时 20ms，假设业务计算耗时 5ms，那么总过耗时 25ms，所以如果能把 4 次 DB 交互优化为 2 次，那么总耗时就可以优化到 15ms，即总体性能提高了 40%。所以应用如果对时延敏感的话，可以把 DB 部署在 SSD 硬盘，相比于 SCSI 磁盘，前者的 RT 会小很多。