Spanner 佇列情境和範例

本文提供架構模式和程式碼範例，說明如何使用 Spanner 佇列處理常見的訊息傳送情境。您可以使用這些模式，在交易提交後觸發非同步工作、安排延遲或週期性工作、使用頻外儲存空間管理大型訊息酬載、協調多事件工作流程，以及檢查點或延長長期執行的背景工作租約。

僅需處理一次，且最多確認一次

如要進一步瞭解僅需處理一次作業和最多確認一次作業的各種考量事項和解決方案，請參閱「僅需處理一次作業和最多確認一次作業」頁面。

在交易提交後執行工作

如要在交易提交後執行工作，請在同一筆交易中將訊息傳送至佇列。

舉例來說，新使用者註冊時會觸發歡迎電子郵件：

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": "[e-mail]"}'
);

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": "[e-mail]"}' 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()。

例如產生報表：

1. 收件者會收到 RECEIVE_ReportQueue() 傳送的訊息。
2. 開始生成報表。
3. 每 5 秒呼叫 SELECT * FROM RENEWLEASE_ReportQueue([leaseToken]) ，並使用獨立的執行緒或常式。
4. 完成後，請確認訊息並儲存報表。

或者，如果您有需要「最多處理一次」的長時間工作，或租用時間較長，請執行下列操作：

1. 在收到目前佇列訊息時確認 (DELETE 或 ACK)。在同一筆交易中，重新將新的佇列訊息加入佇列，並將傳送時間戳記設為未來的時間，超過處理作業所需的時間。
2. 繼續處理，並在完成後確認新加入佇列的訊息。

這種做法的優點是不需持續延長租約，且訊息不會重新傳送，直到未來時間到來 (涵蓋當機情況)。如果初始確認成功，即可實現最多一次的處理作業。

檢查長時間執行的工作

Spanner 佇列可管理耗時數分鐘到數小時的工作，而不只是快速工作。對於這些長時間執行的工作，請使用下列方法：

1. 在外部儲存中繼資料：使用頻外儲存空間來保存工作的詳細資料和狀態。
2. 定期設定檢查點：如要從當機狀態復原，且不會遺失太多進度，工作應定期儲存自身狀態。
3. 使用建議的檢查點模式：檢查點的最佳做法是原子式確認 (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.

等待多個事件，再繼續操作

如要在繼續作業前等待多個事件 (例如聯結作業)，請使用表格追蹤狀態，並使用佇列觸發檢查。

舉例來說，訂單出貨需要庫存和付款：

1. 使用 InventoryStatus 和 PaymentStatus 建立 Orders 資料表。
2. 確認庫存後，請更新 Orders 並傳送訊息至 OrderCheckQueue。
3. 確認付款後，請更新 Orders 並傳送訊息至 OrderCheckQueue。
4. OrderCheckQueue 的接收器會檢查 Orders 表格。如果兩個狀態都已確認，系統會繼續出貨並確認訊息。如果沒有，系統可能會重新排隊，稍後再檢查，或執行其他邏輯。

定期執行動作

如要定期執行動作，請使用定期排程模式。接收者確認訊息，並傳送排定在下一個間隔傳送的新訊息。

舉例來說，每小時資料匯總：

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 佇列，包括最佳做法和監控。
• 瞭解僅須處理一次的作業和最多確認一次。
• 使用佇列的精細存取控管設定存取控管。