Dataflow (Apache Beam) を大驥?のメッセージをバッチ処理する場合に使える

　　Pub/Sub→Dataflow→処理

StreamingPull API を使用するとアプリとの間で永続的な藹??方向接続が維持され、Pub/Sub でメッセージが利用可能になるとすぐ縺? pullされる。1 つ縺? pull リクエスト縺? 1 つ縺? pull レスポンスが返る通常縺? 単項 Pull と觸??較すると、高スループット・臀??レイテンシ。必要なメッセージを残す処理をしたりも？GCP側の問題であっても通信が切れた場合は別サーバに軆??縺?なおすためmodifyAckDeadlineも切れ再配信されるバグがある

＋＋＋

メッセージ縺?TTL (Time-To-Live) はメッセージ臀??持期間(message retention duration) に臀??存

メッセージが TTLを超えると、自動的に削除され、Subscriberが藹??信できな縺?なる

ackDeadlineSeconds (デフォルト縺?10秒、最螟?600秒) を超えたACKのメッセージは再配信されますが、TTL期限を超えた場合は觸??える

#TTLを最螟?7日間に設定

gcloud pubsub subscriptions update my-subscription message-retention-duration=604800s

DLQ (Dead Letter Queue)
Subscriberが指定回謨?(最螟?100回) メッセージ縺?ACKを行繧?なかった場合に、メッセージを隔離する仕組縺?

DLQもサブスクなので期間やTTL設藹??方觸??は同じ

#DLQ topic 作成

gcloud pubsub topics create my-dlq-topic

#5回失敗したらDLQ縺?

gcloud pubsub subscriptions update my-subscription dead-letter-topic=projects/my-project/topics/my-diq-topic max-delivery-attempts=5

#DLQ subsc作成

gcloud pubsub subscriptions create my-diq-subscription--topic-my-diq-topic

#サブスクの詳細確認

gcloud pubsub subscriptions describe my-diq-subscription

#DLQメッセージの確認、-auto-ackも付けられるが、

gcloud pubsub subscriptions pull my-dlq-subscription -limit=10

#base64のでコードが必要

echo "SGVsbG8gd29ybGQ=" | base64-decode

#ack-idによりackを返しDLQメッセージを削髯?

gcloud pubsub subscriptions acknowledge my-diq-subscription--ack-ids=Y3g49NfFY

モニタリン繧? > アラートポリシーから新しいアラートを作成し

pubsub.subscription.outstanding_messages を監鐔??対象に選択し、閾値を設藹??するとよい

#DLQ メッセージの再処理をfunctionsに設定 (トピックに入れなおす)

from google.cloud import pubsub_v1

publisher = pubsub_v1.PublisherClient()

topic_path = publisher.topic_path("my-project", "my-topic")

def republish_message(message):

    future = publisher.publish(topic_path, message.data)

    print(f"Republished message ID: {future.result()}")

subscriber = pubsub_v1.SubscriberClient()

subscription_path = subscriber.subscription_path("my-project", "my-dlq-subscription")

def callback(message):

    print(f"Received message: {message.data}")

    republish_message(message)

    message.ack()

subscriber.subscribe(subscription_path, callback=callback)