背景信息

以上問題的具體答案，請參考下文的PushConsumer和SimpleConsumer。

功能概述

如上圖所示，云消息隊列 RocketMQ 版的消費者處理消息時主要經(jīng)過以下階段：消息獲取—>消息處理—>消費狀態(tài)提交。

針對以上幾個階段，云消息隊列 RocketMQ 版提供了不同的消費者類型：PushConsumer和SimpleConsumer。這兩種類型的消費者通過不同的實現(xiàn)方式和接口可滿足您在不同業(yè)務(wù)場景下的消費需求。具體差異如下：

PushConsumer

PushConsumers是一種高度封裝的消費者類型，消費消息僅通過消費監(jiān)聽器處理業(yè)務(wù)并返回消費結(jié)果。消息的獲取、消費狀態(tài)提交以及消費重試都通過云消息隊列 RocketMQ 版的客戶端SDK完成。

使用方式

PushConsumer的使用方式比較固定，在消費者初始化時注冊一個消費監(jiān)聽器，并在消費監(jiān)聽器內(nèi)部實現(xiàn)消息處理邏輯。由云消息隊列 RocketMQ 版的SDK在后臺完成消息獲取、觸發(fā)監(jiān)聽器調(diào)用以及進行消息重試處理。

//消費示例：使用PushConsumer消費普通消息。
ClientServiceProvider provider = ClientServiceProvider.loadService();
        String topic = "Your Topic";
        FilterExpression filterExpression = new FilterExpression("Your Filter Tag", FilterExpressionType.TAG);
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                //設(shè)置消費者分組。
                .setConsumerGroup("Your ConsumerGroup")
                //設(shè)置接入點。
                .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your Endpoint").build())
                //設(shè)置預(yù)綁定的訂閱關(guān)系。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                //設(shè)置消費監(jiān)聽器。
                .setMessageListener(new MessageListener() {
                    @Override
                    public ConsumeResult consume(MessageView messageView) {
                        //消費消息并返回處理結(jié)果。
                        return ConsumeResult.SUCCESS;
                    }
                })
                .build();
                

內(nèi)部原理

在PushConsumer類型中，消息的實時處理能力是基于SDK內(nèi)部的典型Reactor線程模型實現(xiàn)的。如下圖所示，SDK內(nèi)置了一個長輪詢線程，先將消息異步拉取到SDK內(nèi)置的緩存隊列中，再分別提交到消費線程中，觸發(fā)監(jiān)聽器執(zhí)行本地消費邏輯。

可靠性重試

PushConsumer消費者類型中，客戶端SDK和消費邏輯的唯一邊界是消費監(jiān)聽器接口。客戶端SDK嚴格按照監(jiān)聽器的返回結(jié)果判斷消息是否消費成功，并做可靠性重試。所有消息必須以同步方式進行消費處理，并在監(jiān)聽器接口結(jié)束時返回調(diào)用結(jié)果，不允許再做異步化分發(fā)。消息重試具體信息，請參見PushConsumer消費重試策略。

順序性保障

基于云消息隊列 RocketMQ 版順序消息的定義，如果消費者分組設(shè)置了順序消費模式，則PushConsumer在觸發(fā)消費監(jiān)聽器時，嚴格遵循消息的先后順序。業(yè)務(wù)處理邏輯無感知即可保證消息的消費順序。

適用場景

SimpleConsumer

SimpleConsumer是一種接口原子型的消費者類型，消息的獲取、消費狀態(tài)提交以及消費重試都是通過消費者業(yè)務(wù)邏輯主動發(fā)起調(diào)用完成。

使用方式

SimpleConsumer的使用涉及多個接口調(diào)用，由業(yè)務(wù)邏輯按需調(diào)用接口獲取消息，然后分發(fā)給業(yè)務(wù)線程處理消息，最后按照處理的結(jié)果調(diào)用提交接口，返回服務(wù)端當前消息的處理結(jié)果。示例如下：

 //消費示例：使用SimpleConsumer消費普通消息，主動獲取消息處理并提交。
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        String topic = "Your Topic";
        FilterExpression filterExpression = new FilterExpression("Your Filter Tag", FilterExpressionType.TAG);

        SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
                //設(shè)置消費者分組。
                .setConsumerGroup("Your ConsumerGroup")
                //設(shè)置接入點。
                .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your Endpoint").build())
                //設(shè)置預(yù)綁定的訂閱關(guān)系。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();
        List<MessageView> messageViewList = null;
        try {
            //SimpleConsumer需要主動獲取消息，并處理。
            messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                //消費處理完成后，需要主動調(diào)用ACK提交消費結(jié)果。
                try {
                    simpleConsumer.ack(messageView);
                } catch (ClientException e) {
                    e.printStackTrace();
                }
            });
        } catch (ClientException e) {
            //如果遇到系統(tǒng)流控等原因造成拉取失敗，需要重新發(fā)起獲取消息請求。
            e.printStackTrace();
        }

SimpleConsumer主要涉及以下幾個接口行為：

可靠性重試

SimpleConsumer消費者類型中，客戶端SDK和服務(wù)端通過ReceiveMessage和AckMessage接口通信。客戶端SDK如果處理消息成功則調(diào)用AckMessage接口；如果處理失敗只需要不回復(fù)ACK響應(yīng)，即可在定義的消費不可見時間到達后觸發(fā)消費重試流程。更多信息，請參見SimpleConsumer消費重試策略。

順序性保障

基于云消息隊列 RocketMQ 版順序消息的定義，SimpleConsumer在處理順序消息時，會按照消息存儲的先后順序獲取消息。即需要保持順序的一組消息中，如果前面的消息未處理完成，則無法獲取到后面的消息。

適用場景

使用建議

PushConsumer合理控制消費耗時，避免無限阻塞

對于PushConsumer消費類型，需要嚴格控制消息的消費耗時，盡量避免出現(xiàn)消息處理超時導(dǎo)致消息重復(fù)。如果業(yè)務(wù)經(jīng)常會出現(xiàn)一些預(yù)期外的長時間耗時的消息，建議使用SimpleConsumer，并設(shè)置好消費不可見時間。