DefaultMQPushConsumer 的流量控制
PullMessageService根据偏移量从broker拉取一批消息后会首先放入 ProccessQueue处理队列，然后将这些消息消费任务提交到线程池之后返回。
ProcessQueue对象里主要的内容是一个 TreeMap 和一个读写锁。 TreeMap 里以 Message Queue 的 Offset作为 Key，以消息内容的引用为 Value，保存了 所有从 MessageQueue 获取到，但是还未被处理的消息; 读写锁控制着多个线程对 TreeMap 对象的并发访问。
有了ProcessQueue 对象，流量控制就方便和灵活多了，客户端在每次 Pull请求前会做几个判断，分别取但还未处理的消息个数、消息总大小、Offset的跨度，任何一个值超过设定的大小就隔一段时间再拉取消息，从而达到流量控制的目的。此外 ProcessQueue 还可以辅助实现顺序消费的逻辑。
代码实现在DefaultMQPushConsumerImpl#pullMessage方法里。将消息任务提交到线程池是在这个方法里的PullCallback回调里进行的,其中有ConsumeMessageConcurrentlyService,ConsumeMessageOrderlyService这两个实现，ConsumeRequest就是封装后的任务。
DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest消息拉取方式
- 经过队列负载机制后，会分配给当前消费者一些队列，注意一个消费组可以订阅多个主题，正如上面 pullRequestQueue 中所示，topic_test、topic_test2 这两个主题都分配了一个队列。
- 轮流从 pullRequestQueue 中取出一个 PullRequest 对象，根据该对象中的拉取偏移量向 Broker 发起拉取请求，默认拉取 32 条，可通过 pullBatchSize 参数进行改变，该方法不仅会返回消息列表，还会返更改 PullRequest 对象中的下一次拉取的偏移量。
- 接收到 Broker 返回的消息后，会首先放入 ProccessQueue（处理队列），该队列的内部结构为 TreeMap，key 存放的是消息在消息消费队列（consumequeue）中的偏移量，而 value 为具体的消息对象。
- 然后将拉取到的消息提交到消费组内部的线程池，并立即返回，并将 PullRequest 对象放入到 pullRequestQueue 中，然后取出下一个 PullRequest 对象继续重复消息拉取的流程，从这里可以看出，消息拉取与消息消费是不同的线程。
- 消息消费组线程池处理完一条消息后，会将消息从 ProccessQueue 中删除，然后会向 Broker 汇报消息消费进度，以便下次重启时能从上一次消费的位置开始消费。
消息消费进度提交
通过上面拉取消息流程我们知道，消息消费组线程池在处理完一条消息后，会将消息从 ProccessQueue 中移除，并向 Broker 汇报消息消费进度。
那请大家思考一下下面这个问题：
例如现在处理队列中有 5 条消息，并且是线程池并发消费，那如果消息偏移量为 3 的消息（3:msg3）先于偏移量为 0、1、2 的消息处理完，那向 Broker 如何汇报消息消费进度呢？
如果提交 msg3 的偏移量为消息消费进度，那汇报完毕后如果消费者发生内存溢出等问题导致 JVM 异常退出，msg1 的消息还未处理，然后重启消费者，由于消息消费进度文件中存储的是 msg3 的消息偏移量，会继续从 msg3 开始消费，会造成消息丢失。
RocketMQ 采取的方式是处理完 msg3 之后，会将 msg3 从消息处理队列中移除，但在向 Broker 汇报消息消费进度时是取 ProceeQueue 中最小的偏移量为消息消费进度，即汇报的消息消费进度是 0。
但如果出现上图这种情况，也就是0,1,3都已消费完且从ProcessQueue移除，那如果我们汇报的消费进度为2，如果发生内存溢出等异常情况，消费者重新启动，会继续从消息偏移量为 2 的消息开始消费，msg3 就会被消费多次(msg3会重新被broker里取出)，故RocketMQ 不保证消息重复消费，所以消费的幂等是需要业务方进行保证的。
再看下提交消费进度的流程图：
为了减少消费者与 Broker 的网络交互，提高性能，提交消息消费进度时会首先存入到本地缓存表中，然后定时上报到 Broker，同样 Broker 也会首先存储本地缓存表，然后定时刷写到磁盘。
