Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

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": "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])。
完成后，确认消息并存储报告。

或者，如果您有需要最多处理一次的长时间运行的工作，或者租用时间较长，请执行以下操作：

在到达时确认（DELETE 或 ACK）当前队列消息。在同一事务中，重新将新队列消息排入队列，并设置一个未来的传送时间戳，该时间戳应超过处理所需的时间。
继续处理，并在完成后确认新入队的邮件。

此方法的优点在于，无需不断延长租期，并且在到达未来时间之前不会重新传送消息（这涵盖了崩溃情况）。如果初始确认成功，则实现“最多一次”处理。

为长时间运行的工作设置检查点

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 表。如果两个状态都已确认，则继续发货并确认消息。如果不是，则可能会重新排队以供稍后检查，或执行其他逻辑。

定期执行操作

如需定期执行某项操作，请使用周期性调度模式。接收方确认收到消息，并发送一条计划在下一个时间间隔发送的新消息。

例如，每小时数据汇总：

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 队列场景和示例 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

一次性处理和最多一次确认

在交易提交后执行工作

GoogleSQL

PostgreSQL

GoogleSQL

PostgreSQL

处理长时间运行的工作

为长时间运行的工作设置检查点

安排在未来的特定时间执行工作

GoogleSQL

PostgreSQL

处理大型消息载荷

GoogleSQL

PostgreSQL

等待多个事件，然后再继续

定期执行操作

GoogleSQL

PostgreSQL

Java

Go

Python

Node.js

后续步骤

Spanner 队列场景和示例