ApsaraMQ for RocketMQ:コンシューマタイプ

背景情報

上記の問題に対する回答については、「PushConsumer」および「SimpleConsumer」をご参照ください。

機能の概要

上の図は、ApsaraMQ for RocketMQでのコンシューマーによるメッセージ消費には、メッセージの受信、メッセージの処理、および消費ステータスのコミットの段階が含まれることを示しています。

ApsaraMQ for RocketMQは、PushConsumerとSimpleConsumerのコンシューマータイプを提供します。 2つのタイプのコンシューマは、異なる実装方法とAPI操作を提供することにより、さまざまなメッセージ消費シナリオに適しています。 次の表に、2つのタイプのコンシューマーの違いを示します。

PushConsumer

PushConsumerは、高度なカプセル化を提供するコンシューマータイプです。 メッセージの消費と消費結果の送信は、メッセージリスナーのみを使用して処理されます。 メッセージの取得、消費ステータスの送信、および消費の再試行は、ApsaraMQ for RocketMQクライアントSDKを使用して完了します。

使用状況

PushConsumerは固定的に使用されます。 メッセージリスナーは、コンシューマーが初期化されるときにPushConsumerコンシューマーに登録され、メッセージ処理ロジックがメッセージリスナーに実装されます。 メッセージの取得、リスナー呼び出しのトリガー、およびメッセージの再試行は、ApsaraMQ for RocketMQ SDKを使用して処理されます。

// Message consumption example: Use a PushConsumer consumer to consume messages. 
ClientServiceProvider provider = ClientServiceProvider.loadService();
        String topic = "Your Topic";
        FilterExpression filterExpression = new FilterExpression("Your Filter Tag", FilterExpressionType.TAG);
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                // Configure consumer groups. 
                .setConsumerGroup("Your ConsumerGroup")
                // Specify the access point. 
                .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your Endpoint").build())
                // Specify the pre-bound subscriptions. 
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // Set the message listener. 
                .setMessageListener(new MessageListener() {
                    @Override
                    public ConsumeResult consume(MessageView messageView) {
                        // Consume the messages and return the consumption result. 
                        return ConsumeResult.SUCCESS;
                    }
                })
                .build();
                

働くメカニズム

PushConsumerの場合、リアルタイムメッセージ処理はSDKの一般的なReactorスレッドモデルに基づいています。 SDKには、メッセージをプルしてキューに保存する長いポーリングスレッドが組み込まれています。 次に、メッセージはキューから個々のメッセージ消費スレッドに配信されます。 メッセージリスナーは、メッセージ消費ロジックに基づいて動作します。 次の図は、PushConsumerコンシューマーのメッセージ消費プロセスを示しています。

信頼性の再試行

PushConsumerの場合、クライアントSDKと消費ロジックユニット間の通信は、メッセージリスナーのみを使用して実装されます。 クライアントSDKは、メッセージリスナーによって返された結果に基づいてメッセージが消費されたかどうかをチェックし、メッセージの信頼性を確保するために消費再試行ロジックに基づいて再試行を実行します。 すべてのメッセージは同期的に消費されなければならない。 リスナー操作の呼び出しが終了すると、消費結果が返されます。 非同期配布は許可されていません。 メッセージの再試行の詳細については、「プッシュモードで使用されるメッセージの再試行ポリシー」をご参照ください。

確実なメッセージの順序

ApsaraMQ For RocketMQの注文メッセージの場合、注文メッセージの消費がコンシューマーグループに設定されている場合、PushConsumerコンシューマーは消費順序でメッセージを消費します。 PushConsumerの消費者がメッセージを消費すると、個々のビジネスアプリケーションがビジネスロジックで消費順序を定義する必要なく、消費順序が保証されます。

シナリオ

SimpleConsumer

SimpleConsumerは、メッセージ処理のアトミック操作をサポートするコンシューマタイプです。 このようなタイプのコンシューマは、メッセージを取得し、消費ステータスをコミットし、ビジネスロジックに基づいてメッセージの再試行を実行する操作を呼び出します。

使用状況

SimpleConsumerには複数のAPI操作が含まれます。 対応する操作は、必要に応じて呼び出され、メッセージを取得して処理のためにビジネススレッドに配信します。 次に、コミット動作が呼び出されて、メッセージ処理結果をコミットする。 サンプルコード：

 // Consumption example: When a SimpleConsumer consumer consumes normal messages, the consumer obtains messages and commits message consumption results. 
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        String topic = "Your Topic";
        FilterExpression filterExpression = new FilterExpression("Your Filter Tag", FilterExpressionType.TAG);

        SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
                // Configure consumer groups. 
                .setConsumerGroup("Your ConsumerGroup")
                // Specify the access point. 
                .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("Your Endpoint").build())
                // Specify the pre-bound subscriptions. 
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();
        List<MessageView> messageViewList = null;
        try {
            // A SimpleConsumer consumer must obtain and process messages. 
            messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                // After consumption is complete, the consumer must invoke ACK to submit the consumption result. 
                try {
                    simpleConsumer.ack(messageView);
                } catch (ClientException e) {
                    e.printStackTrace();
                }
            });
        } catch (ClientException e) {
            // If the pull fails due to system traffic throttling or other reasons, the consumer must re-initiate the request to obtain the message. 
            e.printStackTrace();
        }

次の表に、SimpleConsumerに提供されるAPI操作を示します。

信頼性の再試行

SimpleConsumerがメッセージを消費すると、クライアントSDKとApsaraMQ forRocketMQサーバー間の通信は、ReceiveMessageおよびAckMessage操作を使用して実装されます。 クライアントSDKがメッセージを正常に処理すると、AckMessage操作が呼び出されます。 メッセージの処理に失敗した場合、指定されたメッセージの不可視期間が経過した後、メッセージ再試行メカニズムをトリガーするためのackメッセージは返されません。 詳細については、「シンプルモードで使用されるメッセージの再試行ポリシー」をご参照ください。

確実なメッセージの順序

ApsaraMQ for RocketMQでは、SimpleConsumerコンシューマは、順序付けられたメッセージを格納された順序で取得します。 順序付けられたメッセージの詳細については、「順序付けられたメッセージ」をご参照ください。 順序メッセージのセット内のメッセージが完全に処理されない場合、順序メッセージのセット内の次のメッセージを取得することはできません。

シナリオ

使用上の注意

PushConsumerの適切な消費期間制限を指定します。

メッセージが長時間処理されないように、PushConsumerのメッセージ消費期間を制限することをお勧めします。 メッセージを長時間処理すると、メッセージ処理のタイムアウトによりメッセージが重複し、次のメッセージが継続的に消費されるのを待つことがあります。 メッセージが過度に長い期間にわたって頻繁に処理される場合は、SimpleConsumerを使用し、ビジネス要件に基づいて適切なメッセージの不可視期間を指定することを推奨します。