RocketMQ——Consumer篇：向Broker同步消费进度的定时任务

每隔5秒调用一次MQClientInstance.persistAllConsumerOffset()方法将消费进度向Broker同步。遍历MQClientInstance.consumerTable: ConcurrentHashMap<String/*group */, MQConsumerInner>变量。对于PushConsumer端和PullConsumer端，处理逻辑是一样的,以DefaultMQPushConsumerImpl为例，调用DefaultMQPushConsumerImpl.persistConsumerOffset()方法。

1、获取DefaultMQPushConsumerImpl.rebalanceImpl变量的processQueueTable:ConcurrentHashMap<MessageQueue, ProcessQueue>变量值，取该变量的key值集合，即MessageQueue集合；以该集合为参数调用OffsetStore.persistAll(Set<MessageQueue> mqs)方法；

2、若消息模式是广播（BROADCASTING），即DefaultMQPushConsumerImpl.offsetStore变量初始化为LocalFileOffsetStore对象，在此调用LocalFileOffsetStore.persistAll(Set<MessageQueue> mqs)方法，在此方法中遍历LocalFileOffsetStore.offsetTable:ConcurrentHashMap<MessageQueue,AtomicLong>变量，将包含在第1步的MessageQueue集合中的MessageQueue对象的消费进度持久化到consumerOffset.json物理文件中；

3、若消息模式为集群模式，即DefaultMQPushConsumerImpl.offsetStore变量初始化为RemoteBrokerOffsetStore对象，在此调用RemoteBrokerOffsetStore.persistAll(Set<MessageQueue> mqs)方法，在此方法中遍历RemoteBrokerOffsetStore.offsetTable:ConcurrentHashMap <MessageQueue,AtomicLong>变量；对于包含在第1步的MessageQueue集合中的MessageQueue对象，调用updateConsumeOffsetToBroker(MessageQueuemq, long offset)方法向Broker发送UPDATE_CONSUMER_OFFSET请求码的消费进度信息；

 

消费进度(offset)

消费进度是指，当一个consumer group里的consumer在消费某个queue里的消息时，equeue是通过记录消费位置(offset)来知道当前消费到哪里了。以便该consumer重启后继续从该位置开始消费。比如一个topic有4个queue，一个consumer group有4个consumer，则每个consumer分配到一个queue，然后每个consumer分别消费自己的queue里的消息。equeue会分别记录每个consumer对其queue的消费进度，从而保证每个consumer重启后知道下次从哪里开始继续消费。实际上，也许下次重启后不是由该consumer消费该queue了，而是由group里的其他consumer消费了，这样也没关系，因为我们已经记录了这个queue的消费位置了。所以可以看出，消费位置和consumer其实无关，消费位置完全是queue的一个属性，用来记录当前被消费到哪里了。另外一点很重要的是，一个topic可以被多个consumer group里的consumer订阅。不同consumer group里的consumer即便是消费同一个topic下的同一个queue，那消费进度也是分开存储的。也就是说，不同的consumer group内的consumer的消费完全隔离，彼此不受影响。还有一点就是，对于集群消费和广播消费，消费进度持久化的地方是不同的，集群消费的消费进度是放在broker，也就是消息队列服务器上的，而广播消费的消费进度是存储在consumer本地磁盘上的。之所以这样设计是因为，对于集群消费，由于一个queue的消费者可能会更换，因为consumer group下的consumer数量可能会增加或减少，然后就会重新计算每个consumer该消费的queue是哪些，这个能理解的把？所以，当出现一个queue的consumer变动的时候，新的consumer如何知道该从哪里开始消费这个queue呢？如果这个queue的消费进度是存储在前一个consumer服务器上的，那就很难拿到这个消费进度了，因为有可能那个服务器已经挂了，或者下架了，都有可能。而因为broker对于所有的consumer总是在服务的，所以，在集群消费的情况下，被订阅的topic的queue的消费位置是存储在broker上的，存储的时候按照不同的consumer group做隔离，以确保不同的consumer group下的consumer的消费进度互补影响。然后，对于广播消费，由于不会出现一个queue的consumer会变动的情况，所以我们没必要让broker来保存消费位置，所以是保存在consumer自己的服务器上。

 

http://blog.csdn.net/meilong_whpu/article/details/77065587

https://www.cc362.com/content/Npz3glwzPQ.html