Pub/Sub

Pub/Subとは

Pub/Sub（Publish/Subscribe）は、Google Cloud Platform (GCP) が提供するフルマネージドのメッセージングサービスです。アプリケーション間で非同期にメッセージを送受信するためのプラットフォームを提供し、分散システムやマイクロサービスアーキテクチャにおけるコンポーネント間の疎結合な通信を実現します。Pub/Subは、高いスケーラビリティと信頼性を備え、グローバルに分散したアプリケーション間でもリアルタイムなメッセージ配信を可能にします。

Pub/Subの主な特徴

• 非同期メッセージング： 送信者（パブリッシャー）と受信者（サブスクライバー）の間で非同期通信を実現
• 高いスケーラビリティ： 毎秒数百万のメッセージを処理可能で、需要に応じて自動的にスケール
• グローバル可用性： 世界中のどのリージョンからでもアクセス可能なグローバルサービス
• 少なくとも1回の配信保証： メッセージが確実に配信されることを保証
• 順序保証オプション： 特定の条件下でメッセージの順序を保証するオプション
• プッシュとプル配信： サブスクライバーがメッセージをプルする方式と、Pub/Subがエンドポイントにプッシュする方式の両方をサポート
• メッセージフィルタリング： 属性に基づいてメッセージをフィルタリングする機能
• デッドレターポリシー： 処理できないメッセージを特別なトピックに転送する機能
• セキュリティ機能： IAMによるアクセス制御、転送中および保存時の暗号化

Pub/Subのアーキテクチャ

Pub/Subは以下の主要コンポーネントで構成されています：

Pub/Subの使用方法

Pub/Subは、Google Cloud Console、gcloud CLIツール、クライアントライブラリ、またはRESTful APIを使用して利用できます。

トピックの作成

gcloud pubsub topics create my-topic

gcloud pubsub topics update my-topic \
    --message-retention-duration=7d

サブスクリプションの作成

gcloud pubsub subscriptions create my-pull-subscription \
    --topic=my-topic \
    --ack-deadline=60 \
    --message-retention-duration=7d

gcloud pubsub subscriptions create my-push-subscription \
    --topic=my-topic \
    --push-endpoint=https://example.com/push \
    --ack-deadline=60

メッセージの発行

gcloud pubsub topics publish my-topic \
    --message="Hello, World!" \
    --attribute="key1=value1,key2=value2"

メッセージの受信（プル型）

gcloud pubsub subscriptions pull my-pull-subscription \
    --auto-ack \
    --limit=10

クライアントライブラリを使用したPub/Subの操作

const {PubSub} = require('@google-cloud/pubsub');
const pubsub = new PubSub();

async function publishMessage() {
  const topicName = 'my-topic';
  const data = Buffer.from('Hello, World!');
  
  const messageId = await pubsub.topic(topicName).publish(data, {
    key1: 'value1',
    key2: 'value2'
  });
  
  console.log(`Message ${messageId} published.`);
  return messageId;
}

publishMessage();

const {PubSub} = require('@google-cloud/pubsub');
const pubsub = new PubSub();

function listenForMessages() {
  const subscriptionName = 'my-pull-subscription';
  const subscription = pubsub.subscription(subscriptionName);
  
  const messageHandler = message => {
    console.log(`Received message ${message.id}:`);
    console.log(`Data: ${message.data.toString()}`);
    console.log(`Attributes: ${JSON.stringify(message.attributes)}`);
    
    // メッセージを確認（アクノリッジ）
    message.ack();
  };
  
  subscription.on('message', messageHandler);
  
  console.log(`Listening for messages on ${subscriptionName}...`);
}

listenForMessages();

from google.cloud import pubsub_v1

def publish_message():
    publisher = pubsub_v1.PublisherClient()
    topic_path = publisher.topic_path('my-project', 'my-topic')
    
    data = 'Hello, World!'.encode('utf-8')
    attributes = {
        'key1': 'value1',
        'key2': 'value2'
    }
    
    future = publisher.publish(topic_path, data, **attributes)
    message_id = future.result()
    
    print(f'Message {message_id} published.')
    return message_id

publish_message()

from google.cloud import pubsub_v1

def listen_for_messages():
    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path('my-project', 'my-pull-subscription')
    
    def callback(message):
        print(f'Received message {message.message_id}:')
        print(f'Data: {message.data.decode("utf-8")}')
        print(f'Attributes: {message.attributes}')
        
        # メッセージを確認（アクノリッジ）
        message.ack()
    
    streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
    print(f'Listening for messages on {subscription_path}...')
    
    try:
        streaming_pull_future.result()
    except KeyboardInterrupt:
        streaming_pull_future.cancel()
        subscriber.close()

listen_for_messages()

Pub/Subの高度な機能

メッセージフィルタリング

属性に基づいてメッセージをフィルタリングするサブスクリプションを作成できます：

gcloud pubsub subscriptions create my-filtered-subscription \
    --topic=my-topic \
    --filter="attributes.key1 = \"value1\" AND attributes.key2 = \"value2\""

デッドレターポリシー

処理できないメッセージを特別なトピックに転送するデッドレターポリシーを設定できます：

gcloud pubsub subscriptions create my-subscription \
    --topic=my-topic \
    --dead-letter-topic=my-dead-letter-topic \
    --max-delivery-attempts=5

順序保証

メッセージの順序を保証するには、順序キーを使用します：

const {PubSub} = require('@google-cloud/pubsub');
const pubsub = new PubSub();

async function publishOrderedMessages() {
  const topicName = 'my-topic';
  const topic = pubsub.topic(topicName, {
    messageOrdering: true
  });
  
  // 同じ順序キーを持つメッセージは順序通りに配信される
  const orderingKey = 'user-123';
  
  for (let i = 1; i <= 10; i++) {
    const data = Buffer.from(`Message ${i}`);
    const messageId = await topic.publish(data, {}, orderingKey);
    console.log(`Message ${messageId} published with ordering key: ${orderingKey}`);
  }
}

publishOrderedMessages();

スキーマ

メッセージの構造を定義し、検証するためのスキーマを使用できます：

# Avroスキーマの作成
gcloud pubsub schemas create my-schema \
    --type=AVRO \
    --definition="{\"type\":\"record\",\"name\":\"MyRecord\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"value\",\"type\":\"int\"}]}"

# スキーマをトピックに関連付け
gcloud pubsub topics create my-schema-topic \
    --schema=my-schema \
    --message-encoding=JSON

Pub/Subのセキュリティ

Pub/Subは、メッセージングシステムのセキュリティを確保するための様々な機能を提供しています：

アクセス制御

IAMを使用して、トピックとサブスクリプションへのアクセスを制御できます：

# トピックへの発行権限を付与
gcloud pubsub topics add-iam-policy-binding my-topic \
    --member=serviceAccount:my-service-account@my-project.iam.gserviceaccount.com \
    --role=roles/pubsub.publisher

# サブスクリプションへのサブスクライブ権限を付与
gcloud pubsub subscriptions add-iam-policy-binding my-subscription \
    --member=serviceAccount:my-service-account@my-project.iam.gserviceaccount.com \
    --role=roles/pubsub.subscriber

暗号化

Pub/Subは、保存時と転送時のデータを自動的に暗号化します。さらに、カスタマー管理の暗号化キー（CMEK）を使用することもできます：

gcloud pubsub topics create my-cmek-topic \
    --topic-encryption-key=projects/my-project/locations/global/keyRings/my-keyring/cryptoKeys/my-key

VPCサービスコントロール

VPCサービスコントロールを使用して、Pub/Subへのアクセスをセキュリティ境界内に制限できます。

Pub/Subのモニタリングと可観測性

Pub/SubはGoogle Cloud Monitoringと統合されており、トピックとサブスクリプションの状態を監視できます：

主なモニタリングメトリクス

• 発行レート： トピックに発行されるメッセージの数
• 配信レート： サブスクリプションに配信されるメッセージの数
• 未確認メッセージ数： サブスクリプションで未確認のメッセージの数
• 最古の未確認メッセージの経過時間： 最も古い未確認メッセージが発行されてからの時間
• バックログサイズ： サブスクリプションのバックログのサイズ

ログの分析

Cloud Loggingを使用して、Pub/Subの操作ログを分析できます：

gcloud logging read 'resource.type="pubsub_topic" AND resource.labels.topic_id="my-topic"'

Pub/Subのユースケース

Pub/Subは以下のようなユースケースに適しています：

• イベント駆動型アーキテクチャ： イベントの発生に応じて処理を実行するアーキテクチャ
• マイクロサービス間の通信： 疎結合なマイクロサービス間のメッセージング
• ストリーム処理： リアルタイムデータストリームの処理
• データパイプライン： 異なるシステム間のデータ転送
• IoTデータ収集： IoTデバイスからのデータの収集と処理
• 分析データの取り込み： 分析システムへのデータの取り込み
• 非同期ワークフロー： 長時間実行される処理を非同期に実行
• ファンアウト処理： 1つのイベントに対して複数の処理を並行して実行

Pub/Subとの統合サービス

Pub/Subは他のGoogle Cloudサービスと統合して、より強力なソリューションを構築できます：

Cloud Functionsとの統合例

gcloud functions deploy process-pubsub-message \
    --runtime nodejs14 \
    --trigger-topic=my-topic \
    --entry-point=processMessage

Pub/Subのベストプラクティス

• 適切なアクノリッジメント： メッセージを処理した後に必ずアクノリッジする
• べき等性の確保： メッセージが複数回配信される可能性を考慮して、べき等性を確保する
• バッチ処理： 可能な場合はメッセージをバッチで処理してスループットを向上させる
• 適切なアクノリッジメント期限： メッセージ処理時間に合わせてアクノリッジメント期限を設定する
• デッドレターポリシーの使用： 処理できないメッセージを処理するためのデッドレタートピックを設定する
• メッセージサイズの最適化： メッセージサイズを小さく保ち、大きなデータは外部ストレージに保存する
• フィルタリングの活用： クライアント側でのフィルタリングを減らすためにサーバー側のフィルタリングを使用する
• モニタリングとアラートの設定： バックログサイズや未確認メッセージ数などの重要なメトリクスを監視する

まとめ

Pub/Sub は、Google Cloud Platform上での非同期メッセージングを簡素化するフルマネージドサービスです。高いスケーラビリティと信頼性を備え、分散システムやマイクロサービスアーキテクチャにおけるコンポーネント間の疎結合な通信を実現します。

Pub/Subの主な利点は以下の通りです：

• 高いスケーラビリティと可用性によるメッセージングの信頼性
• グローバルな可用性によるどこからでもアクセス可能なメッセージング
• プッシュとプル配信モデルの柔軟性
• メッセージフィルタリングによる効率的なメッセージ処理
• 他のGCPサービスとのシームレスな統合
• 強力なセキュリティ機能によるメッセージの保護

Pub/Subを効果的に活用するには、適切なトピックとサブスクリプションの設計、べき等性の確保、バッチ処理の最適化、モニタリング戦略の実装が重要です。また、他のGCPサービスと組み合わせることで、より強力なイベント駆動型アーキテクチャを構築することができます。