RocketMQ的消费模式和消息流控

消费模式

1、Push模式--PushConsumer

消费端SDK内置了一个长轮询线程，先将消息异步拉取到SDK内置的缓存队列中，再分别提交到消费线程中，触发监听器执行本地消费逻辑。PushConsumer 消费者类型中，客户端SDK和消费逻辑的唯一边界是消费监听器接口。客户端SDK严格按照监听器的返回结果判断消息是否消费成功，并做可靠性重试。所有消息必须以同步方式进行消费处理，并在监听器接口结束时返回调用结果，不允许再做异步化分发。

适用场景

PushConsumer严格限制了消息同步处理及每条消息的处理超时时间，适用于以下场景：

• 消息处理时间可预估：如果不确定消息处理耗时，经常有预期之外的长时间耗时的消息，PushConsumer的可靠性保证会频繁触发消息重试机制造成大量重复消息。
• 无异步化、高级定制场景：PushConsumer限制了消费逻辑的线程模型，由客户端SDK内部按最大吞吐量触发消息处理。该模型开发逻辑简单，但是不允许使用异步化和自定义处理流程。

2、Pull模式--SimpleConsumer

Pull模式中，整个消息获取过程由消费端自由控制，通过simpleConsumer.receive()从broker获取消息。消费端可根据自身业务处理能力来控制从broker获取消息的速度。缺点是消费者需要不断轮询获取消息，在broker暂时没有消息时，消费端处于空转状态，存在性能问题，并且消息获取间隔长又容易导致消费不及时。

适用场景

SimpleConsumer提供原子接口，用于消息获取和提交消费结果，相对于PushConsumer方式更加灵活。SimpleConsumer适用于以下场景：

• 消息处理时长不可控：如果消息处理时长无法预估，经常有长时间耗时的消息处理情况。建议使用SimpleConsumer消费类型，可以在消费时自定义消息的预估处理时长，若实际业务中预估的消息处理时长不符合预期，也可以通过接口提前修改。
• 需要异步化、批量消费等高级定制场景：SimpleConsumer在SDK内部没有复杂的线程封装，完全由业务逻辑自由定制，可以实现异步分发、批量消费等高级定制场景。
• 需要自定义消费速率：SimpleConsumer是由业务逻辑主动调用接口获取消息，因此可以自由调整获取消息的频率，自定义控制消费速率。

消息流控

生产端流控

在生产端要达到消息生产速度的控制。一、可以实现基础设施中的限流框架，如Sentinel，进行消息生产接口的流量控制。二、可在broker端配置消息产生速度，如果达到阈值，直接返回异常。在消息积压特别严重时，可通过该手段，防止broker崩溃。

消费端流控

消费端流控主要是如何提高消息的消费能力，大概有的思路，消费线程数量、消费应用数量、业务处理能力，批量处理。

在消费端，消费能力不够，容易造成服务端消息积压。对于PushConsumer消费类型，需要严格控制消息的消费耗时，尽量避免出现消息处理超时导致消息重复。如果业务经常会出现一些预期外的长时间耗时的消息，建议使用SimpleConsumer，并设置好消费不可见时间。