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)