はじめに

🎄 本記事は ZOZO Advent Calendar 2024 シリーズ４の３日目です

Cloud Pub/Subを用いる際、構築するシステム要件によっては、
メッセージの処理順序を考慮する必要があります。
例えば、メッセージ内の主キーを用いてSubscriberでDBのレコードを挿入するか更新するかなどの処理です。

詳しくは順序指定の概要で紹介しますが、
Pub/Subはデフォルトではメッセージの受信の順序を保証していません。

「どれぐらいの頻度でメッセージの順序が逆転するんだろう？」と思ったので、調べてみます。

本記事では、単一Publisherで単一Subscriberの順序指定をしない場合において、
メッセージの処理が逆転する確率を調査してみます。

順序指定の概要

本記事の調査を行うために、そもそもPub/Subの順序指定における概要を調査します。

Pub/SubのSubscriberのデフォルトでは、メッセージの受信の順序を保証していません。
メッセージを順序に受信したい場合は、Order messagesの機能を用いることで実現できます。

Publisher側の設定

サブスクライバークライアントでメッセージを順に受信するには、順序指定キーを使用してメッセージをパブリッシュするようにパブリッシャークライアントを構成する必要があります。

引用通り、メッセージのPublish時に順序指定キーを使用します。
個人的に特筆すべき事項は以下です。

リージョン間の順序指定
- Pub/SubにはGlobal endpointsとLocational endpointsの2種類のエンドポイントが存在します（参照）
- デフォルトではGlobal endpointsが使用されますが、順序指定を行う際は同一のリージョンPublishする必要があるので、Locational endpointsを使用する必要があります
複数のPublisher間の順序指定
- Publisherが複数ある場合、同一の順序指定キーを使用した同時Publishは避けるべきだそうです
- つまり順序指定キーに対して、順次処理する1つのPublisherが望ましいですが、
  複数のPublisherで構成する場合は、順序指定キー別にロックするような構成が必要です

Subscriber側の設定

Subscriptionを作成する際に、[順序指定キーを使用してメッセージの順序を指定する] を設定します。

また下記のような考慮事項があります。

メッセージの再配信
- Pub/Subでは、少なくとも1回の配信を保証されますので、重複したメッセージの配信の可能性があります
- 重複したメッセージが再配信されると、確認済みメッセージを含む後続のメッセージも再配信されます
確認応答の遅延とデッドレタートピック
- 確認応答処理が遅延すると、他の順序指定キーのメッセージの配信が遅れることがあります
- 即時に確認応答処理ができない場合は、デッドレタートピックを用いるなどが推奨されいます
Dataflow との統合
- Dataflowを用いる際は、Dataflow独自の順序指定を保証した処理を行います
- よって、順序指定オプションは指定しないようにしてください（パフォーマンスの低下の可能性がある）

調査

手順

最初にTopicの作成を行う。

gcloud pubsub topics create research_reversal_topic

次に、Subscription。

gcloud pubsub subscriptions create research_reversal_sub --topic=research_reversal_topic

ここで、 --enable-message-ordering の指定はしない。

次にPublisherの実装。

from google.cloud import pubsub_v1

project_id = "your-project-id"
topic_id = "research_reversal_topic"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

MESSAGE_SIZE = ここを変えながら複数試行する

for n in range(1, MESSAGE_SIZE + 1):
    data_str = f"{n}"
    data = data_str.encode("utf-8")
    future = publisher.publish(topic_path, data)
    while future.done() is False:
        pass
    print(f"Published message: {data_str}")

print(f"Published messages to {topic_path}.")

確実に順序通りPublishを行いため、Publishが終わるまでの待ち処理を入れている。

最後に、Subscriberの実装。

from google.api_core import retry
from google.cloud import pubsub_v1

project_id = "your-project-id"
subscription_id = "research_reversal_sub"

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_id)

received_messages = []


def pull_messages():
    global received_messages  # validate_reverse で使うため
    try:
        with subscriber:
            while True:
                response = subscriber.pull(
                    request={"subscription": subscription_path, "max_messages": 1},
                    timeout=10,
                    retry=retry.Retry(deadline=300),
                )

                ack_ids = []
                for received_message in response.received_messages:
                    data = received_message.message.data.decode("utf-8")
                    print(f"Received message: {data}")

                    current_message_number = int(data)
                    received_messages.append(current_message_number)

                    ack_ids.append(received_message.ack_id)

                if ack_ids:
                    subscriber.acknowledge(
                        request={"subscription": subscription_path, "ack_ids": ack_ids}
                    )

                print(
                    f"Received and acknowledged {len(response.received_messages)} messages from {subscription_path}."
                )
    except Exception as e:
        print(f"Error while pulling messages: {e}")
    finally:
        validate_reverse()


def validate_reverse():
    reverse_count = 0

    # 隣接する要素を比較して逆転を検出
    for i in range(1, len(received_messages)):
        if received_messages[i] < received_messages[i - 1]:
            reverse_count += 1
            print(
                f"Reversal detected: {received_messages[i - 1]} -> {received_messages[i]}"
            )

    print(f"Total reversals detected: {reverse_count}")
    return reverse_count


# メッセージを引き出す処理を実行
pull_messages()

結果

上記コードを実行した結果が下記。

実験No	メッセージ数	逆転回数	割合(%)
1	100	0	0.00
2	100	2	2.00
3	100	1	1.00
4	100	7	7.00
5	100	1	1.00
6	1,000	25	2.50
7	1,000	19	1.90
8	1,000	20	2.00
9	1,000	18	1.80
10	1,000	17	1.70
11	10,000	141	1.41
12	10,000	176	1.76
13	10,000	156	1.56
14	10,000	176	1.76
15	10,000	153	1.53

またある程度、Subscriptionにメッセージが溜まった場合、比較的にPublish時間が最新のものが処理されることが確認できた。
例えば、メッセージ数が10,000の実験において、1~5,000番台は順調にpullしていたが、次に急に8,000番台のメッセージがpullされたなどである。（時間幅が大きい）

おわりに

割と順序通りではなく、驚き...😭
また、本記事ではSubscriberをすでに起動した状態で待機して、メッセージのPublishを行ったが、先にPublishして、Subscriptionにメッセージが溜まった状態でSubscriberを起動すると、逆転回数が多くなる傾向がみられた。
本記事では、簡単な処理を行うSubscriberだったので応答確認処理（ack）が早かったが、この辺りも実運用を考えると更に逆転が増えそう...😭

そもそもこの調査のモチベーションとして、
公式ドキュメントより、順序指定を行う際は考慮事項が多かったり、順序指定キーの粒度が荒かったりするとパフォーマンスが悪く¹なるようなので、
「めっちゃ稀に逆転起きて、逆転する時間幅が狭い」なら許容できるケースあるのでは？と思い、そのある程度の基準値を調べようとおもったのがきっかけ。

しかし調査結果的に、逆転回数が多く、また逆転の時間幅が大きいので、順序を考慮する要件がある場合は、順序指定を用いた方が良さそう...

https://christina04.hatenablog.com/entry/gcp-cloud-pubsub-ordering-key-concern ↩