RocketMQ的消费者在消费消息时可能会遇到各种异常,常见的有:
- 消息校验失败:消费的消息不符合预期格式,导致无法解析。
- 序列化失败:消费的消息无法反序列化成Java对象。
- 业务处理失败:消费逻辑在处理消息时抛出异常。
- 网络异常:Consumer和Broker之间网络中断,消息消费失败。
- Broker异常:Broker故障导致消息消费失败。
为了保证消息的可靠消费,RocketMQ提供了以下手段处理消费异常:
- 自动重试:Consumer在消费失败后会在一定时间重试消费,直到成功或者超过最大重试次数。重试时隔和最大重试次数可以配置。
- 回溯消费:Consumer在重试期间如果发现有新消息产生,会先消费新消息,然后继续回溯消费失败的消息。这样可以避免消息积压。
- 灵活的重试策略:支持多种重试策略,如每次固定睡眠时间重试、随着重试次数增长睡眠时间递增、随机睡眠时间重试等。可以选用合适的策略以兼顾消费速度和消息积压。
- 消息重发:如果Consumer在最大重试次数内仍然消费失败,Broker会定期扫描被标记为消费失败的消息,并将其重新发送给其他在线Consumer消费。
- 补偿消费:同一条消息会被多个Consumer消费,Consumer在处理失败后会将补偿关联的消息重新进行消费,即“两阶段提交”模式。适用于广播型消费场景。
- 消息过滤:可根据Tag、属性等过滤条件过滤不符合预期格式的“脏消息”,避免消费异常。
- 自定义错误处理逻辑:Consumer可以实现ConsumeConcurrentlyContext#handleConsumeException方法来自定义异常处理逻辑。
通过以上机制,RocketMQ能够较好地处理消费异常场景,保证消息的高可靠消费。理解这些机制的原理,可以让我们在使用RocketMQ构建高可靠的消息系统时更加得心应手。