Platform For AI:キューサービスのサブスクリプションとプッシュ

コンシューマー

コンシューマーとは、キューサービスからデータをサブスクライブするクライアントプログラムを指します。クライアントが Watch API を使用してデータを呼び出すと、キューサービス内にコンシューマーオブジェクトが作成されます。API 内のパラメーター (ウィンドウサイズやタグなど) は、コンシューマーの属性として使用されます。コンシューマーのステータスは、Attribute API を通じて確認できます。例：

[OK] Attributes: 
consumers.list.[0] : Id: default_group.u1, Index: 0, Pending: 0, Status: Complete, Idle: 2.091s, Window: 0, Slots: 0, AutoCommit: true
consumers.list.[1] : Id: default_group.u2, Index: 0, Pending: 0, Status: Complete, Idle: 1.124s, Window: 0, Slots: 0, AutoCommit: true
consumers.stats.total : 2

consumers.stats.total はコンシューマーの総数を示します。consumers.list はコンシューマーのリストです。各列の説明は次のとおりです：

パラメーター	説明
Id	コンシューマーの ID。フォーマットは <コンシューマーグループ ID.コンシューマー ID> です。
Index	コンシューマーが消費しているデータのインデックス。
Pending	このコンシューマーが処理中で、まだコミットしていないデータ量。
Status	コンシューマーのステータス。有効値： Running：コンシューマーは実行中です。 Exit：コンシューマーが長期間離脱しており、データは消費されていません。 Complete：コンシューマーが離脱し、データは消費済みです。 Leaving：コンシューマーが短期間離脱しています。
Window	コンシューマーウィンドウのサイズ。プッシュされるデータの最大量です。
Slots	アイドリング状態のウィンドウ数。値が 0 の場合は、ウィンドウが占有されていることを示します。
AutoCommit	データ送信後にデータを自動的にコミットするかどうかを示します。
Tags	コンシューマーのフィルター条件。 説明 Watch API をタグ付きで使用する場合、同じコンシューマーグループ内のコンシューマーが使用するタグは、すべて同じにしてください。

Watch API を使用する際にデータタグを指定すると、次のように Tags 列が追加で表示されます。例：

consumers.list.[0] : Id: ..., Pending: 0, ..., Window: ..., Tags: tags[foo=bar]

この列は、コンシューマーがサブスクライブするデータタグを示します。データは、条件を満たした場合にのみコンシューマーに配信されます。

コンシューマーグループ

コンシューマーグループは、同じフィルター条件でキューサービスをサブスクライブするコンシューマーのコレクションです。同じグループ内のコンシューマーは同じ名前を持つことはできませんが、異なるグループ内のコンシューマーは同じ名前を持つことができます。

同じコンシューマーグループ内では、データは各コンシューマーに均等に分散されます。異なるグループ間では、データは各グループのコンシューマーに並行してプッシュされます。例：

• 同じグループ内のコンシューマーは、異なるデータを受信します。

• 異なるグループ内のコンシューマーは、同じデータを受信します。

Attribute API を通じて、キューサービス内のコンシューマーグループのステータスを確認できます。以下に例を示します：

groups.list.[0] : Id: default_group, Index: 0, Pending: 0, Delivered: 0, Consumers: 1
groups.list.[1] : Id: group, Index: 0, Pending: 0, Delivered: 1, Consumers: 0

groups.list はコンシューマーグループのリストです。各列の説明は次のとおりです：

パラメーター	説明
Id	コンシューマーグループの ID。
Index	現在のコンシューマーグループが消費しているデータのインデックス。グループ内の全コンシューマーのうち、最も大きいインデックス値を指します。
Pending	このコンシューマーグループが処理中で、まだコミットしていないデータ量。
Delivered	プッシュされたメッセージの数。
Consumers	コンシューマーグループ内のコンシューマー数。

コンシューマーグループの数に制限はありませんが、自動的にクリーンアップされることはありません。作成後もその状態は保持されます。

コンシューマーとコンシューマーグループの使用

コンシューマーとコンシューマーグループは、Watch API の HTTP ヘッダー、または SDK でクライアントを初期化する際に宣言できます。HTTP ヘッダーのキーは、Attributes API を使用して取得することもできます。

meta.header.group : X-EAS-QueueService-Gid
meta.header.user : X-EAS-QueueService-Uid

X-EAS-QueueService-Uid と X-EAS-QueueService-Gid を通じて、それぞれコンシューマー ID と参加するコンシューマーグループ ID を宣言します。

コミットとネガティブ

キューサービスは、コミットとネガティブという 2 つの消費方式をサポートしています。どちらもデータのインデックスに対して操作を行いますが、セマンティクスが異なります。

• コミットは、コンシューマーがデータを受信して処理を完了し、次のバッチをプッシュできる状態になったことを示します。

• ネガティブは、コンシューマーがデータを受信したものの処理できないことを示し、キューサービスはエラーコードに基づいて次のバッチをプッシュするかどうかを決定します。ネガティブモードでは、原因とエラーコードをテキストで指定できるため、そのデータを他のコンシューマーにプッシュさせることができます。次の表に、キューサービスが処理できる特別なエラーコードを示します：

  コード	説明
  Shutdown	コンシューマーが終了処理中であり、キューサービスがデータのプッシュを継続しないことを示します。

データリバランス

以下のシナリオでは、データをコミットできない場合があります：

• 予測サービスがローリングアップデートされる際、一部のコンシューマーが終了し、処理中のデータをコミットできなくなる場合。

• 何らかの内部エラーが発生し、コンシューマーがクラッシュする場合。

• コンシューマーが受信したデータを処理できず、ネガティブコミットを実行する場合。

このように処理できないデータは、キューサービスによって他のコンシューマーに再配布されます。このメカニズムを データリバランス と呼びます。データリバランスは、次の場合に実行されます：

• いずれかのコンシューマーが Exit 状態になった場合。

• ウィンドウに空きがあるにもかかわらず、コンシューマーが新しいデータのプッシュを受信しない場合。

キューサービスは、データごとに配信カウンターを維持します。データがリバランスされ再配布されるたびに、このカウンターはインクリメントされます。リバランスの過程で、あるデータの配信カウンターが最大配信回数を超えたことが判明した場合、そのデータは デッドレター として扱われます。キューサービスは、設定されたデッドレターポリシーを実行します。デフォルトでは、データはテイルキューに配信されます。

テイルキュー

テイルキューは、コンシューマーにプッシュされなかったデータ (デッドレターやカスタムコントロールデータなど) を格納するために使用される補助的なキューです。これはキューサービス内のキューインスタンスであり、同じ API を持ちます。各入力キューと出力キューには、それぞれテイルキューがあります。

API を呼び出す際、次の HTTP ヘッダーを追加することで、テイルキューへのアクセスを宣言できます：

X-EAS-QueueService-Access-Rear: true