Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

Spanner キューのシナリオと例

このドキュメントでは、Spanner キューを使用する一般的なメッセージングシナリオのアーキテクチャパターンとコードサンプルについて説明します。これらのパターンを使用すると、トランザクションの commit 後に非同期処理をトリガーしたり、遅延タスクや定期的なタスクのスケジュールを設定したり、帯域外ストレージで大きなメッセージペイロードを管理したり、マルチイベントワークフローを調整したり、長時間実行されるバックグラウンドジョブのチェックポイントを設定したり、リースを延長したりできます。

1 回限りの処理と最大 1 回の確認応答

1 回限りの処理と 1 回限りの確認応答に関するさまざまな考慮事項と解決策については、1 回限りの処理と 1 回限りの確認応答のページで詳しく説明しています。

トランザクションの commit 後に作業を行う

トランザクションの commit 後に処理を実行するには、同じトランザクション内でキューにメッセージを送信します。

たとえば、新規ユーザーの登録によってウェルカムメールがトリガーされます。

GoogleSQL

-- Inside your application transaction:
-- 1. Insert into Users table
INSERT INTO Users (UserId, UserName) VALUES (124, 'New User');

-- 2. Send message to queue to trigger email
INSERT INTO UserTasks (UserId, MessageId, Payload)
VALUES (
  124,
  'welcome-email-id',
  b'{"type": "welcome", "email": "user@example.com"}'
);

PostgreSQL

-- Inside your application transaction:
-- 1. Insert into users table
INSERT INTO users (userid, username) VALUES (124, 'New User');

-- 2. Send message to queue to trigger email
INSERT INTO usertasks (userid, messageid, payload)
VALUES (
  124,
  'welcome-email-id',
  CAST('{"type": "welcome", "email": "user@example.com"}' AS bytea)
);

トランザクションがコミットされると、UserTasks のレシーバーはメッセージをストリーミングし、メールを送信して、メッセージを確認します。

GoogleSQL

-- 1. In the receiver process, stream messages from the queue
SELECT
  UserId,
  MessageId,
  Payload,
  DeliverTime,
  SpannerLeaseExpirationTimestamp,
  SpannerLeaseToken
FROM RECEIVE_UserTasks(max_duration=>'20m');

-- 2. After sending the welcome email, acknowledge the message
DELETE FROM UserTasks
WHERE UserId = 124 AND MessageId = 'welcome-email-id';

PostgreSQL

-- 1. In the receiver process, stream messages from the queue
SELECT
  userid,
  messageid,
  payload,
  deliver_time,
  spanner_lease_expiration_timestamp,
  spanner_lease_token
FROM spanner.receive_usertasks(NULL, NULL, '20m');

-- 2. After sending the welcome email, acknowledge the message
DELETE FROM usertasks
WHERE userid = 124 AND messageid = 'welcome-email-id';

長時間実行される作業を処理する

デフォルトのリース時間（10 秒超）よりも長くかかる可能性がある作業がある場合は、SELECT * FROM RENEWLEASE_QUEUE_NAME() を定期的に呼び出します。

たとえば、レポートを生成する場合:

受信者は RECEIVE_ReportQueue() からメッセージを受け取ります。
レポートの生成を開始します。
5 秒ごとに、別のスレッドまたはルーチンで SELECT * FROM RENEWLEASE_ReportQueue([leaseToken]) を呼び出します。
完了したら、メッセージを確認してレポートを保存します。

また、1 回限りの処理または長いリース時間を必要とする長時間実行の作業がある場合は、次の操作を行います。

到着時に現在のキューメッセージを確認します（DELETE または ACK）。同じトランザクションで、処理にかかる時間を超える将来の配信タイムスタンプを使用して、新しいキューメッセージを再エンキューします。
処理を続行し、完了したら新しくキューに登録されたメッセージを確認応答します。

このアプローチの利点は、リースを継続的に延長する必要がないことと、将来の時刻になるまでメッセージが再配信されないことです（クラッシュに対応）。最初の確認応答が成功すると、1 回限りの処理が実現します。

長時間実行される作業のチェックポイント

Spanner キューは、数分から数時間続くタスクを管理できます。クイックジョブだけでなく、このような長時間実行タスクには、次のアプローチを使用します。

メタデータを外部に保存する: 帯域外ストレージを使用して、タスクの詳細と状態を保持します。
定期的にチェックポイントを作成する: クラッシュから復旧し、進行状況をあまり失わないようにするには、タスクの状態を定期的に保存する必要があります。
推奨のチェックポイントパターンを使用する: チェックポイントの最適な方法は、現在のキューメッセージをアトミックに確認（ACK）し、将来の配信用にスケジュールされた新しいメッセージを送信することです。この新しいメッセージには、更新された状態が含まれているか、その状態が示されています。これにより、別のワーカーへの即時再配信が防止されます。

このパターンでは、完全なチェックポイント設定ができない場合でも、重複する作業を減らすことができます。ただし、このシナリオでは、クラッシュ後にタスクが最初から再開されます。

将来の特定の時間に処理をスケジュールする

将来の特定の時間に作業をスケジュールするには、メッセージの挿入時に DeliverTime 列を設定します。

たとえば、トライアルの有効期限に関するリマインダーは次のようになります。

GoogleSQL

-- 1. Insert into Users table
INSERT INTO Users (UserId, UserName) VALUES (125, 'Trial User');

-- 2. Send message to queue with a future delivery time
INSERT INTO UserTasks (UserId, MessageId, Payload, DeliverTime)
VALUES (125, 'trial-expire-reminder', b'{"type": "reminder"}', TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 29 DAY));

PostgreSQL

-- 1. Insert into users table
INSERT INTO users (userid, username) VALUES (125, 'Trial User');

-- 2. Send message to queue with a future delivery time
INSERT INTO usertasks (userid, messageid, payload, deliver_time)
VALUES (125, 'trial-expire-reminder', CAST('{"type": "reminder"}' AS bytea), CURRENT_TIMESTAMP + INTERVAL '29 DAY');

大規模なメッセージペイロードを処理する

メッセージペイロードが大きい場合は、アウトオブバンドストレージパターンを使用します。大きなペイロードを別のテーブルに保存し、その参照をキューメッセージに配置します。

たとえば、画像処理は次のようになります。

GoogleSQL

-- Schema
CREATE TABLE ImageUploads (
  UserId    INT64 NOT NULL,
  ImageId   STRING(36) NOT NULL,
  ImageData BYTES(MAX),
  Status    STRING(MAX) -- PENDING, PROCESSING, DONE
) PRIMARY KEY (UserId, ImageId),
  INTERLEAVE IN PARENT Users;

CREATE QUEUE ImageProcessingQueue (
  UserId    INT64 NOT NULL,
  ImageId   STRING(36) NOT NULL,
  Payload   BYTES(1) NOT NULL -- Payload can be minimal
) PRIMARY KEY (UserId, ImageId),
  INTERLEAVE IN PARENT ImageUploads ON DELETE CASCADE;

-- Application Logic
-- 1. Upload image, insert into ImageUploads with Status 'PENDING'
-- 2. Send message to ImageProcessingQueue
INSERT INTO ImageProcessingQueue (UserId, ImageId, Payload) VALUES (123, 'image-uuid-1', b'');

-- Receiver for ImageProcessingQueue:
-- 1. Receives message (UserId, ImageId).
-- 2. Reads ImageData from ImageUploads.
-- 3. Processes image.
-- 4. Updates ImageUploads Status to 'DONE'.
-- 5. ACKs the queue message.

PostgreSQL

-- Schema
CREATE TABLE imageuploads (
  userid    bigint NOT NULL,
  imageid   varchar(36) NOT NULL,
  imagedata bytea,
  status    varchar, -- PENDING, PROCESSING, DONE
  PRIMARY KEY (userid, imageid)
) INTERLEAVE IN PARENT users;

CREATE QUEUE imageprocessingqueue (
  userid    bigint NOT NULL,
  imageid   varchar(36) NOT NULL,
  payload   bytea NOT NULL, -- Payload can be minimal
  PRIMARY KEY (userid, imageid)
) INTERLEAVE IN PARENT imageuploads ON DELETE CASCADE;

-- Application Logic
-- 1. Upload image, insert into imageuploads with status 'PENDING'
-- 2. Send message to imageprocessingqueue
INSERT INTO imageprocessingqueue (userid, imageid, payload) VALUES (123, 'image-uuid-1', CAST('' AS bytea));

-- Receiver for imageprocessingqueue:
-- 1. Receives message (userid, imageid).
-- 2. Reads imagedata from imageuploads.
-- 3. Processes image.
-- 4. Updates imageuploads status to 'DONE'.
-- 5. ACKs the queue message.

複数のイベントを待ってから続行する

続行する前に複数のイベントを待機する（結合オペレーションなど）には、テーブルを使用して状態を追跡し、キューを使用してチェックをトリガーします。

たとえば、在庫と支払いを必要とする注文処理は次のようになります。

InventoryStatus と PaymentStatus を使用して Orders テーブルを作成します。
在庫が確認されたら、Orders を更新し、OrderCheckQueue にメッセージを送信します。
支払いが確認されたら、Orders を更新して OrderCheckQueue にメッセージを送信します。
OrderCheckQueue の受信側は Orders テーブルをチェックします。両方のステータスが確認された場合は、配送手続きに進み、メッセージを承認します。そうでない場合は、後でチェックするために再キューに入れるか、他のロジックを実行します。

アクションを定期的に実行する

アクションを定期的に実行するには、定期的なスケジューリングパターンを使用します。受信側はメッセージを確認応答し、次の間隔で送信される新しいメッセージを送信します。

たとえば、1 時間ごとのデータ集計は次のようになります。

GoogleSQL

-- Inside your application transaction:
-- 1. Acknowledge current message
DELETE FROM AggregationQueue
WHERE TaskType = 'hourly-aggregator' AND MessageId = 'current-uuid'
ASSERT_ROWS_MODIFIED 1;

-- 2. Schedule next run 1 hour in the future
INSERT INTO AggregationQueue (TaskType, MessageId, Payload, DeliverTime)
VALUES ('hourly-aggregator', 'next-uuid', b'', TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR));

PostgreSQL

-- Inside your application transaction:
-- 1. Acknowledge current message
DELETE FROM aggregationqueue
WHERE tasktype = 'hourly-aggregator' AND messageid = 'current-uuid'
ASSERT_ROWS_MODIFIED 1;

-- 2. Schedule next run 1 hour in the future
INSERT INTO aggregationqueue (tasktype, messageid, payload, deliver_time)
VALUES ('hourly-aggregator', 'next-uuid', CAST('' AS bytea), CURRENT_TIMESTAMP + INTERVAL '1 HOUR');

または、クライアントライブラリの Ack ミューテーションと Send ミューテーションを使用します。これらの例では、キーとペイロードをカプセル化する Message オブジェクトがあることを前提としています。

Java

// Receiver logic for AggregationQueue
public void process(DatabaseClient dbClient, Message msg) {
  // ... do aggregation ...

  // ACK current message and schedule next run (1 hour from now)
  Instant nextRun = Instant.now().plus(Duration.ofHours(1));
  Mutation ackMutation =
      Mutation.newAckBuilder("AggregationQueue")
          .setKey(msg.getKey()) // Ack
          .build();
  Mutation sendMutation =
      Mutation.newSendBuilder("AggregationQueue")
          .setKey(Key.of("hourly-aggregator", "next-uuid"))
          .setPayload(Value.bytes(ByteArray.copyFrom("")))
          .setDeliveryTime(nextRun) // Schedule next
          .build();
  dbClient.write(Arrays.asList(ackMutation, sendMutation));
}

Go

// Receiver logic for AggregationQueue
func process(msg) {
    // ... do aggregation ...

    // ACK current message and schedule next run
    nextRun := time.Now().Add(1 * time.Hour)
    _, err := client.Apply(ctx, []*spanner.Mutation{
        spanner.Ack("AggregationQueue", msg.Key), // Ack
        spanner.Send("AggregationQueue",
            spanner.Key{"hourly-aggregator", "next-uuid"},
            []byte(""),
            spanner.WithDeliveryTime(nextRun), // Schedule next
        ),
    })
    // ... handle err ...
}

Python

# Receiver logic for AggregationQueue
def process(database: spanner.Database, msg: Message):
  # ... do aggregation ...
  # ACK current message and schedule next run (1 hour from now)
  next_run = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
      hours=1
  )
  with database.batch() as batch:
    batch.ack(
        queue="AggregationQueue",
        key=msg.key,  # Ack
    )
    batch.send(
        queue="AggregationQueue",
        key=("hourly-aggregator", "next-uuid"),
        payload=b"",
        deliver_time=next_run,  # Schedule next
    )

Node.js

/**
 * Receiver logic for AggregationQueue
 * @param {import('@google-cloud/spanner').Database} database
 * @param { { key: Array<string|number>, payload: Buffer } } msg
 */
async function process(database, msg) {
  // ... do aggregation ...
  // ACK current message and schedule next run (1 hour from now)
  const nextRun = new Date(Date.now() + 60 * 60 * 1000);
  await database.runTransactionAsync(async (transaction) => {
    // Ack current message
    transaction.queueAck('AggregationQueue', msg.key);
    // Schedule next run
    transaction.queueSend(
      'AggregationQueue',
      ['hourly-aggregator', 'next-uuid'],
      {
        payload: Buffer.from(''),
        deliverTime: nextRun,
      }
    );
    await transaction.commit();
  });
}

次のステップ

ベストプラクティスとモニタリングを含む、Spanner キューの使用方法について学習する。
1 回限りの処理と最大 1 回の確認応答について学習する。
キューのきめ細かいアクセス制御を使用してアクセス制御を構成します。

Spanner キューのシナリオと例 コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

1 回限りの処理と最大 1 回の確認応答

トランザクションの commit 後に作業を行う

GoogleSQL

PostgreSQL

GoogleSQL

PostgreSQL

長時間実行される作業を処理する

長時間実行される作業のチェックポイント

将来の特定の時間に処理をスケジュールする

GoogleSQL

PostgreSQL

大規模なメッセージ ペイロードを処理する

GoogleSQL

PostgreSQL

複数のイベントを待ってから続行する

アクションを定期的に実行する

GoogleSQL

PostgreSQL

Java

Go

Python

Node.js

次のステップ

Spanner キューのシナリオと例

大規模なメッセージペイロードを処理する