韩城建设局网站,睿达科网络 网站建设,wordpress 弹出 广告,中学网站模板Apache RocketMQ 通过消费位点管理消费进度#xff0c;本文为您介绍 Apache RocketMQ 的消费进度管理机制。
背景信息
Apache RocketMQ 的生产者和消费者在进行消息收发时#xff0c;必然会涉及以下场景#xff0c;消息先生产后订阅或先订阅后生产。这两种场景下#x…Apache RocketMQ 通过消费位点管理消费进度本文为您介绍 Apache RocketMQ 的消费进度管理机制。
背景信息
Apache RocketMQ 的生产者和消费者在进行消息收发时必然会涉及以下场景消息先生产后订阅或先订阅后生产。这两种场景下消费者客户端启动后从哪里开始消费如何标记已消费的消息这些都是由 Apache RocketMQ 的消费进度管理机制来定义的。
通过了解 Apache RocketMQ 的消费进度管理机制可以帮助您解答以下问题 消费者启动后从哪里开始消费消息 消费者每次消费成功后如何标记消息状态确保下次不会再重复处理该消息 某消息被指定消费者消费过一次后如果业务出现异常需要做故障恢复该消息能否被重新消费
消费进度原理
消息位点Offset
参考 Apache RocketMQ 主题和队列的定义消息是按到达服务端的先后顺序存储在指定主题的多个队列中每条消息在队列中都有一个唯一的Long类型坐标这个坐标被定义为消息位点。
任意一个消息队列在逻辑上都是无限存储即消息位点会从0到Long.MAX无限增加。通过主题、队列和位点就可以定位任意一条消息的位置具体关系如下图所示 Apache RocketMQ 定义队列中最早一条消息的位点为最小消息位点MinOffset最新一条消息的位点为最大消息位点MaxOffset。虽然消息队列逻辑上是无限存储但由于服务端物理节点的存储空间有限 Apache RocketMQ 会滚动删除队列中存储最早的消息。因此消息的最小消费位点和最大消费位点会一直递增变化。 消费位点ConsumerOffset
Apache RocketMQ 领域模型为发布订阅模式每个主题的队列都可以被多个消费者分组订阅。若某条消息被某个消费者消费后直接被删除则其他订阅了该主题的消费者将无法消费该消息。
因此Apache RocketMQ 通过消费位点管理消息的消费进度。每条消息被某个消费者消费完成后不会立即在队列中删除Apache RocketMQ 会基于每个消费者分组维护一份消费记录该记录指定消费者分组消费某一个队列时消费过的最新一条消息的位点即消费位点。
当消费者客户端离线又再次重新上线时会严格按照服务端保存的消费进度继续处理消息。如果服务端保存的历史位点信息已过期被删除此时消费位点向前移动至服务端存储的最小位点。
信息
消费位点的保存和恢复是基于 Apache RocketMQ 服务端的存储实现和任何消费者无关。因此 Apache RocketMQ 支持跨消费者的消费进度恢复。
队列中消息位点MinOffset、MaxOffset和每个消费者分组的消费位点ConsumerOffset的关系如下 ConsumerOffset≤MaxOffset 当消费速度和生产速度一致且全部消息都处理完成时最大消息位点和消费位点相同即ConsumerOffsetMaxOffset。 当消费速度较慢小于生产速度时队列中会有部分消息未消费此时消费位点小于最大消息位点即ConsumerOffsetMaxOffset两者之差就是该队列中堆积的消息量。 ConsumerOffset≥MinOffset正常情况下有效的消费位点ConsumerOffset必然大于等于最小消息位点MinOffset。消费位点小于最小消息位点时是无效的相当于消费者要消费的消息已经从队列中删除了是无法消费到的此时服务端会将消费位点强制纠正到合法的消息位点。
消费位点初始值
消费位点初始值指的是消费者分组首次启动消费者消费消息时服务端保存的消费位点的初始值。
Apache RocketMQ 定义消费位点的初始值为消费者首次获取消息时该时刻队列中的最大消息位点。相当于消费者将从队列中最新的消息开始消费。
重置消费位点
若消费者分组的初始消费位点或当前消费位点不符合您的业务预期您可以通过重置消费位点调整您的消费进度。
适用场景 初始消费位点不符合需求因初始消费位点为当前队列的最大消息位点即客户端会直接从最新消息开始消费。若业务上线时需要消费部分历史消息您可以通过重置消费位点功能消费到指定时刻前的消息。 消费堆积快速清理当下游消费系统性能不足或消费速度小于生产速度时会产生大量堆积消息。若这部分堆积消息可以丢弃您可以通过重置消费位点快速将消费位点更新到指定位置绕过这部分堆积的消息减少下游处理压力。 业务回溯纠正处理由于业务消费逻辑出现异常消息被错误处理。若您希望重新消费这些已被处理的消息可以通过重置消费位点快速将消费位点更新到历史指定位置实现消费回溯。
重置功能
Apache RocketMQ 的重置消费位点提供以下能力 重置到队列中的指定位点。 重置到某一时刻对应的消费位点匹配位点时服务端会根据自动匹配到该时刻最接近的消费位点。
使用限制 重置消费位点后消费者将直接从重置后的位点开始消费对于回溯重置类场景重置后的历史消息大多属于存储冷数据可能会造成系统压力上升一般称为冷读现象。因此需要谨慎评估重置消费位点后的影响。建议严格控制重置消费位点接口的调用权限避免无意义、高频次的消费位点重置。 Apache RocketMQ 重置消费位点功能只能重置对消费者可见的消息不能重置定时中、重试等待中的消息。
版本兼容性
关于消费者分组的消费位点初始值不同的服务端版本中定义如下 服务端历史版本4.x/3.x版本消息位点初始值受当前队列消息状态的影响。 服务端5.x版本明确定义消费位点初始值为消费者获取消息时刻队列中的最大消息位点。
因此若您将服务端版本从历史版本升级到最新的5.x版本时需要自行对消费者首次启动时的情况做兼容性判断。
使用建议
严格控制消费位点重置的权限
重置消费位点会给系统带来额外处理压力可能会影响新消息的读写性能。 因此该操作请在适用场景下谨慎执行并提前做好合理性和必要性评估。
