■pubsub

Publisher app → |GCPの壁| Topic(Schema) → Subscription 1や2 |GCPの壁| → Subscriber app
　サブスクライバーappにPull/PushさせるPull/Pushのサブスクリプションをトピックに紐づける設定をしておく

【図解付き】Cloud Pub/Subに概要や使い方についてわかりやすく解説 - KIYONO Engineer Blog (kiyono-co.jp)

アプリで簡単にPubsubにパブリッシュや、サブスクもできるので、アプリ間の連携にPubsubが使える

　• 非同期処理(画像処理とか重めのもの

　• IDの種類 (message id, subscription id, topic id, ack id, project idあたりがアプリでは使われるっぽい

　　※ack idはpull時のみでPushのときはhttpステータスコードが200でackとなる

トピック（メッセージのパブリッシュ先）
　• スキーマ/外部アクセス許可/リテンション/GCS/バックアップの設定がある (Push/Pullの設定はない)

　• パブリッシュ側のベストプラクティス (JWT) 

　　Pub/Sub トピックにパブリッシュするためのベスト プラクティス  |  Pub/Sub ドキュメント  |  Google Cloud

サブスクライバのPushとPull (PushはEndpointが必要、デフォルトはpull)

　GCP - Pub/Sub サービス概要 #GoogleCloud - Qiita

　• at-least-once (少なくとも1回) 配信を提供します

　• 同じ順序指定キーを持ち、同じリージョンに存在している場合は、メッセージの順序指定を有効にできます

　• サブスクライバーが31日間未使用、またはサブスクリプションが未更新の場合、サブスクリプションは期限切れ

　• メッセージ数が多いとpull向き サブスクリプション タイプを選択する  |  Pub/Sub ドキュメント  |  Google Cloud

pushはhttpsが必要?

　push サブスクリプションを作成する  |  Pub/Sub ドキュメント  |  Google Cloud

　• push エンドポイントのサーバーには、認証局が署名した有効な SSL証明書が必要でhttps

　• Cloud run でEvent Arcを設定するとサブスクが自動作成されrunのデフォルトhttpsのURLが使われるが、これはPullよりPushで安定した

　• CronバッチならPullで安定するのでは?大量リクエストはPull向きとある(Pullは失敗処理込みの話かも知れん)

トピックのリテンション:デフォルトなし、最小値:10分、最大値:31日

サブスクのリテンション:デフォルト値:7日、最小值:10分、最大値:7日

　サブスクリプション プロパティ  |  Pub/Sub ドキュメント  |  Google Cloud

pubsub ack期限(Ack Deadline)
　•デフォルト60秒> 設定10分>ack延長で最大1時間まで伸ばせると思われる

　リース管理で確認時間を延長する  |  Pub/Sub ドキュメント  |  Google Cloud

　•exactly onceを設定しなければ期限の延長は保証されない

　•ack期限を過ぎる、あるいはNackを返す場合、メッセージは再配送される

　•ack応答期限の延長は99パーセンタイル(上位1%の値よりも小さい値のうち最大の値)で

　modifyAckDeadlineを返し、延長してもMaxExtension (ack期限を延長 する最大値) 60minまでと思われる

メッセージの再試行を強制するには
　•nack リクエストを送信

　•高レベルのクライアント ライブラリを使用していない場合は、ackDeadlineSeconds を0に設定して modifyAckDeadline リクエストを送信する

exactly once
1 回限りの配信  |  Pub/Sub ドキュメント  |  Google Cloud

　•pullなら設定できる。他には、Cloud Dataflowを組み合わせる(プログラムコードでDataflowを使う感じかり、あるいはmessageについているunique idを利用して、KVS を用いたステート管理をして自前で重複を排除する

　•再配信は、メッセージに対してクライアントによる否定確認応答が行われた場合、または確認応答期限が切れる前にクライアントが確認応答期限を延長しな かった場合のいずれかか原因で発生することがある。

pubsubはトピックにPublishされたメッセージをDataflowに引き継げる

　•Apache Beamのウィンドウ処理とセッション分析とコネクタのエコシスエムがある

　•メッセージ重複の削除ができる

　•pubsub>dataflow>BQやGCS: この流れでログ等をストーリミングで入れ込める

BQサブスクリプション (PubSubはBigQuery Storage Write API を使用してデータを BigQueryテーブルに送信、GCSサブスクもある)

　Langganan BigQuery  |  Dokumentasi Pub/Sub  |  Google Cloud

　BigQuery サブスクリプションの作成  |  Pub/Sub ドキュメント  |  Google Cloud

サブスクライバーApp側のコードでのフロー制御によりちょっと待てよのトラフィック急増対応

　フロー制御を使用して一時的な急増を処理する  |  Pub/Sub ドキュメント  |  Google Cloud

デッドレタートピック (配信試行回数が見れる)やエラーでの再配信

　メッセージ エラーの処理  |  Pub/Sub ドキュメント  |  Google Cloud

　 • Pub/Subサブスクリプションにデッドレタートピックを設定しておくと、一定の回数再送信が失敗したメッセージの宛先がデッドレタートピックに変更され貯められる

メッセージのフィルタ、同時実行制御により多いメッセージに対応

　サブスクリプションからのメッセージをフィルタする  |  Pub/Sub ドキュメント  |  Google Cloud

Pubsubをローカルでエミュレートする

　エミュレータを使用したローカルでのアプリのテスト  |  Pub/Sub ドキュメント  |  Google Cloud

pubsubのスナップショットやリテンション

トピックにリテンションを設定しスナップショット作成> 過去のサブスクしたメッセは見えなさそう

サブスクにリテンションを設定しスナップショット作成> 過去のAckしたメッセは見えなさそう

スナップショットでどう使うのか?

　cloud pubsubで配信済みのメッセージを再送する #PubSub - Qiita

　キューがたまっているときに撮るものと思われる。またシーク時間のポイントを設定する意味がある

　スナップショットとシークを使いこなして特定期間の再実行を行う機能

　　スナップショットで再実行する

　　シークは指定時間か最後のスナップショット以降のサブスク再実行(実際pushでrunが再実行された)

Pubsubにどんなメッセージが入ってきているか確認する方法

　pull形式ならAckしなければpullボタンで拾い見れる (トピックでパブリッシュしてサブスクでPull し見る)

　トラブルシュートはログを見るかデッドレタートピックかGCSバックアップを見る?

デッドレターキュー(ドロップしたものの確認と救済?)

　サブスクでDLQのONしデッドレタートピックを設定し転送する>GCSにもバックアップできる

　DLTでメッセージ(実行済みOR未実行)の再生

データ形式:スキーマを使うか、スキーマなしならdataで取得できる

　トピックのスキーマを作成する  |  Pub/Sub ドキュメント  |  Google Cloud

　Cloud Pub/Subの概要とPythonでの実践 - case-kの備忘録

from google cloud import pubsub_v1

from avro.io import DatumReader, BinaryDecoder

from avro schema import Parse

project_id="your-project-id"

subscription id="your-subscription-id"

subscriber pubsub_v1.SubscriberClient()

subscription_path = subscriber.subscription_path(project_id, subscription_id)

avro_schema = Parse("""

{

"type": "record",

"name": "Avro".

"fields": [

{

"name": "ProductName",

"type": "string",

"default":""

},

{

"name": "SKU",

"type": "int",

"default": 0

}

}

def callback(message):

subscriber.subscribe(subscription_path, callback=callback)

def callback(message):