ApsaraMQ for Kafka:Best practices for consumers

頻繁なリバランスの防止

リバランスは、パーティションが再割り当てされている間、消費を一時停止します。頻繁なリバランスはスループットを妨げ、処理レイテンシーを増加させます。これらはコンシューマーハートビートがタイムアウトしたときに発生します。

頻繁なリバランスを防ぐには、関連するパラメーターを調整するか、消費レートを向上させてください。詳細なトラブルシューティング手順については、「コンシューマークライアントでリバランスが頻繁に発生するのはなぜですか？」をご参照ください。

推奨パーティション数

ApsaraMQ for Kafka コンソールでのデフォルトのパーティション数は 12 であり、ほとんどのワークロードで機能します。スケーリング時には、次のガイドラインに従ってください。

1つのグループが複数のトピックをサブスクライブ

単一のコンシューマーグループは複数のトピックをサブスクライブできます。サブスクライブされたすべてのトピックからのメッセージは、グループ内のコンシューマーに均等に分散されます。

String topicStr = kafkaProperties.getProperty("topic");
String[] topics = topicStr.split(",");
for (String topic: topics) {
    subscribedTopics.add(topic.trim());
}
consumer.subscribe(subscribedTopics);

複数のグループが1つのトピックをサブスクライブ

複数のコンシューマーグループが同じトピックを独立してサブスクライブできます。各グループはすべてのメッセージを受信し、グループは互いに影響を与えることなく独立して動作します。

このパターンは、異なるアプリケーションが同じデータストリームを独立して処理する必要がある場合に使用します。たとえば、リアルタイム分析用のグループと、データアーカイブ用の別のグループなどです。

アプリケーションごとに1つのグループを保持

各アプリケーションには専用のコンシューマーグループを使用してください。単一のアプリケーションが異なる消費ロジックを実行する必要がある場合は、異なる group.id 値を持つ個別の構成ファイル (例: kafka1.properties および kafka2.properties) を作成します。

グループ内のサブスクリプションの一致

同じグループ内のすべてのコンシューマーは、同じトピックセットをサブスクライブする必要があります。サブスクリプションの不一致はトラブルシューティングを複雑にし、予期しないリバランス動作につながる可能性があります。

自動コミットと手動コミット

ApsaraMQ for Kafka は、コンシューマオフセットをコミットするための2つのパラメーターを提供します。

自動コミットが有効になっている場合、クライアントは各ポーリングの前に最後のコミットからの経過時間を確認します。経過時間が auto.commit.interval.ms を超えると、現在のオフセットをコミットします。

オフセットの手動コミット

オフセットを手動でコミットするには、enable.auto.commit を false に設定し、各バッチの処理後に commit(offsets) 関数を呼び出します。

オフセットリセット動作

コンシューマオフセットは、次の2つの状況でリセットされます。

• コミットされたオフセットが存在しない場合 -- たとえば、コンシューマーが初めてブローカーに接続するとき。

• コミットされたオフセットが無効な場合 -- たとえば、パーティションの最大オフセットが10であるにもかかわらず、コンシューマーがオフセット11から読み取ろうとするとき。

Java クライアントでのリセット動作は、auto.offset.reset で制御します。

べき等な消費の実装

ご利用のアプリケーションが重複に敏感な場合 (たとえば、金融トランザクションや注文処理など) は、アプリケーションレベルで重複排除を実装してください。

1. 生成時に各メッセージに一意キーを付与します (たとえば、トランザクションID)。

2. 消費されたメッセージを処理する前に、そのキーが既に処理されているかどうかを確認します。

3. 既に処理されたキーを持つメッセージはスキップします。

ご利用のアプリケーションが時折の重複を許容する場合 (たとえば、メトリック集約やログ取り込みなど) は、このステップをスキップしてください。

蓄積の一般的な原因

消費スループットの向上

コンシューマーを追加します。 追加のコンシューマーインスタンス (コンシューマーあたり1スレッド) を起動するか、より多くのコンシューマープロセスをデプロイします。アクティブなコンシューマーの数はパーティション数を超えることはできません。超過分はアイドル状態になります。

処理にスレッドプールを使用します。 ポーリングと処理を分離して作業を並列化します。

1. 固定数のワーカースレッドを持つスレッドプールを定義します。

2. メッセージのバッチをポーリングします。

3. 各メッセージをスレッドプールに送信して並行処理します。

4. バッチ内のすべてのタスクが完了した後、次のバッチをポーリングします。

このパターンにより、ポーリングループの応答性を維持しつつ、複数のスレッドに処理作業を分散させることができます。