Details and best practices of RocketMQ's consumer types

In RocketMQ 5.0, the concept of client type, especially consumer type, is emphasized. There are three different types of consumers in RocketMQ, namely PushConsumer, SimpleConsumer and PullConsumer. Different consumer types correspond to different business scenarios.

Consumer Type Overview

This article will also talk about different types of consumers. Before introducing different message types, let's clarify a common workflow among different RocketMQ consumers: in consumers, messages arriving at the client are obtained by the client actively requesting the server and suspending long polling. In order to ensure the timeliness of message arrival, the client needs to constantly send requests to the server (whether the request needs to be initiated by the client actively depends on the specific client type). Once a new qualified message arrives at the server, the client will request to leave. Finally, the server records the message processing results according to the different processing results of the client.

In addition, PushConsumer and SimpleConsumer also have a concept of ConsumerGroup, which is equivalent to the common identity of a group of consumers with the same subscription relationship. The server will also record the corresponding consumption progress according to the ConsumerGroup. The message consumers in the same ConsumerGroup will consume all the messages that meet the requirements of the current subscription group, instead of consuming independently. Compared with PullConsumer, PushConsumer and SimpleConsumer are more suitable for business integration scenarios. The consumption status and progress are managed by the server, which is relatively lighter and simpler.

To put it simply:

• PushConsumer: a fully hosted consumer type. Users only need to register a message listener, and messages matching the corresponding subscription relationship will call the corresponding consumption method. It is the most common consumer type for business integration.

• SimpleConsumer: a consumer type that decouples message consumption and progress synchronization. Users can independently accept messages from the server and confirm a single message. Like PushConsumer, the consumption progress is managed by the server, which is applicable to business scenarios where users need to control the consumption rate independently.

• PullConsumer: A type of consumer managed by the flow processing framework. Users receive messages according to the queue (the smallest logical unit of the topic) and can choose to submit consumption sites automatically or manually.

PushConsumer

PushConsumer is currently the most widely used consumer of RocketMQ. The user only needs to register the corresponding Listener after confirming the subscription relationship. After a message that conforms to the corresponding subscription relationship is sent by the producer, the consumer's Listener interface will also be called immediately. At this time, the user needs to implement the corresponding business logic in the Listener.

Users need to return ConsumeResult.SUCCESS or ConsumeResult.FAILURE according to their own business processing results. When the user returns ConsumeResult.SUCCESS, the message is considered as successful consumption; When the user returns ConsumeResult. FAILURE, the service side will consider the consumption failure and will perform a backoff retry of the message. The backoff retry of the message means that before the message is successfully consumed, the current message will be delivered to the user's registered MessageListener for many times until the consumption is successful, and the time interval between the two consumption is consistent with the backoff rule.

In particular, each ConsumerGroup will have a maximum consumption setting. If the current message consumption exceeds this setting, the message will not be delivered again, but will be delivered to the dead letter queue instead. This consumption number will automatically increase each time the message is delivered to the MessageListener. For example, if the maximum consumption number of a message is 1, the message will only be consumed once, regardless of whether the message is returned for consumption success or consumption failure.

Application Scenarios and Best Practices

PushConsumer is a nearly fully hosted consumer. The meaning of hosting here is that users do not need to care about the receipt of messages, but only about the consumption process of messages. All other logic is encapsulated in the implementation of Push consumers. Users only need to return different consumption results according to each received message. Therefore, it is also the most popular consumer type.

In most scenarios, the user should quickly process the consumption logic and return the success of consumption, and should not block the consumption logic for a long time. For heavy consumption logic, it is recommended to submit the consumption status first, and then process the message asynchronously.

In fact, in the implementation of push consumers, in order to ensure the timeliness of message consumption, messages will be pulled by the client in advance for subsequent consumption, so there is a cache of the pulled message size in the client. In order to prevent the client memory leakage caused by too many cached messages, client parameters are reserved in advance for users to set themselves.

In SimpleConsumer, users need to pull messages themselves through the SimpleConsumer # receive interface, and then process the pulled messages differently according to their own business logic processing results. SimpleConsumer # receive also accepts messages from the server through long polling. The specific long polling time can be set using SimpleConsumerBuilder # setAwaitDuration.

In SimpleConsumer, the user needs to set a time window (or an invisible time window for messages received through this interface) through SimpleConsumer # receive. The time window starts counting from the time the user receives the message. During this time, the message will not be delivered to the consumer repeatedly, but beyond this time window, The message will be delivered again. In this process, the number of message consumption will also increase. Similar to PushConsumer, once the number of consumption times exceeds the maximum number of times of ConsumerGroup, it will not be re invested.

Compared with PushConsumer, SimpleConsumer users can independently control the rhythm of receiving messages. SimpleConsumer # receive will pull qualified messages from the server for the current subscription relationship. In fact, each message receiving request of SimpleConsumer is initiated one by one according to the specific topic partition. There may be many actual topic partitions. Therefore, in order to ensure the timeliness of message receiving, it is recommended to improve the concurrency of SimpleConsumer # receive to a certain extent by integrating its own business processing capabilities.

After receiving the message, the user can choose to use ack or changeInvisibleDuration for the message. The former means to confirm the message to the server, similar to the successful consumption in PushConsumer, while changeInvisibleDuration means to delay the visible time of the current message, that is, the server needs to deliver the message to the client after the current period of time. It is worth noting that the message redelivery here also needs to follow the limit of the maximum consumption times of the ConsumerGroup. That is, once the maximum consumption times of the message exceed the maximum consumption times (the consumption times will automatically increase each time the message reaches the visible time), the message will not be delivered again and will enter the dead letter queue instead. For example:

• Perform an ack, which means that the message consumption is confirmed successfully and the consumption progress is synchronized by the server.

• Change InvisibleDuration,

1) If the message has exceeded the maximum consumption times of the current ConsumerGroup, the message will be delivered to the dead letter queue subsequently

2) If the message does not exceed the maximum consumption times of the current ConsumerGroup, and the request is initiated before the last message visible time, the modification is successful; otherwise, the modification fails.

Application Scenarios and Best Practices

In PushConsumer, messages are delivered to the MessageListener for processing. In SimpleConsumer, users can get a batch of messages at the same time. The maximum number of messages in each batch is also determined by SimpleConsumer # receive. In some IO intensive applications, it will be a more convenient choice. At this point, users can get a batch of messages each time and process them centrally to improve consumption speed.

PullConsumer

PullConsumer is also a consumer type that RocketMQ has always supported. The new PullConsumer API in RocketMQ 5.0 is still evolving. Please look forward to it. The PullConsumer below will use the existing LitePullConsumer in 4.0 for discussion, which is also the currently recommended method.

Introduction

In RocketMQ, messages are sent or received through queues. A Topic consists of several queues. Messages are stored one by one in the form of queues. Messages in the same queue have different bits, and the size of the bits increases with the time the message reaches the server, Essentially, the consumption progress of different ConsumerGroups on the server is the bit information in the queue. The client synchronizes its consumption progress to the server, which is essentially the bit synchronization of messages.

The concept of queue is completely exposed to users in PullConsumer. The user can set a route listener for the topic he or she cares about to sense the change of the queue and assign the queue to the current consumer. When the user uses LitePullConsumer # poll, he or she will try to obtain the messages in the queue that have been assign. If LitePullConsumer # setAutoCommit is set, the site will be submitted automatically once the message reaches the client. Otherwise, you need to use the LitePullConsumer # commitSync interface to submit manually.

Application Scenarios and Best Practices

In PullConsumer, users have the absolute autonomy to manage message sites, and can manage consumption progress by themselves. This is the most essential difference from PushConsumer and SimpleConsumer, which also makes PullConsumer widely used in flow computing scenarios where both consumption rate and consumption progress need to be controlled independently. In more cases, PullConsumer is integrated with specific stream computing frameworks.