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(
    max_batch_size=>NULL, priority=>NULL, max_duration=>'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 队列以最短延迟时间传送消息。不过，当大量消息从独立客户端持续到达时，单独处理每条消息会产生较高的事务开销。尝试手动查询或扫描队列表以批量处理消息可能会导致范围锁定争用、中止率升高和额外的读取成本。

如需实现无争用的高吞吐量批处理，请应用时间批处理模式。发送者将消息的 DeliverTime 与未来的离散时间窗口对齐（例如，向上舍入到最接近的 10 秒边界）。由于独立发送方会计算出相同的未来时间戳，因此 Spanner 往往会将来自同一分块的消息分组在一起，并在 RECEIVE_QUEUE_NAME() 表值函数中的 max_batch_size 允许的情况下，以单个批次的形式传送这些消息；如果 max_batch_size 小于要传送的消息数量，则以多个批次的形式传送这些消息。

可以使用以下公式计算交付时间戳：

$$ \text{DeliverTime} = \text{RoundDown}(\text{CurrentTime}, \text{FixedDelay}) + \text{FixedDelay} $$

例如，如果时间窗口为 10 秒，那么在 09:05:00 到 09:05:09.999 之间入队的全部消息都会收到 09:05:10 的 DeliverTime：

GoogleSQL

-- Calculate delivery time rounded to the next 10-second interval.
-- Use DIV(..., 10) * 10 to perform the RoundDown in SQL:
INSERT INTO OrderProcessingQueue (OrderId, DeliverTime, Payload)
VALUES (
  'order-101',
  TIMESTAMP_SECONDS(DIV(UNIX_SECONDS(CURRENT_TIMESTAMP()), 10) * 10 + 10),
  b'{"item": "book", "qty": 1}'
);

PostgreSQL

-- Calculate delivery time rounded to the next 10-second interval:
INSERT INTO orderprocessingqueue (orderid, deliver_time, payload)
VALUES (
  'order-101',
  to_timestamp((floor(extract(epoch from CURRENT_TIMESTAMP) / 10) * 10) + 10),
  CAST('{"item": "book", "qty": 1}' AS bytea)
);

接收方随后使用 max_batch_size 分批拉取这些同时发送的消息：

GoogleSQL

SELECT OrderId, Payload, DeliverTime, SpannerLeaseToken
FROM RECEIVE_OrderProcessingQueue(max_duration=>'20m', max_batch_size=>50);

PostgreSQL

SELECT orderid, payload, deliver_time, spanner_lease_token
FROM spanner.receive_orderprocessingqueue(
    max_batch_size=>50, priority=>NULL, max_duration=>'20m');

如果您同时发送数百万条消息，将所有消息精确对齐到同一秒可能会导致处理量突然飙升。为了均匀分配工作，同时仍按实体对消息进行批处理，请根据唯一标识符（例如 TIMESTAMP_ADD(deliver_time, INTERVAL MOD(ABS(FARM_FINGERPRINT(CAST(UserId AS STRING))), 10) SECOND)）向计算中添加偏移量。

将数据修改与处理分离（脏标记模式）

在高吞吐量事务型应用中，直接在面向用户的事务中执行复杂的重新计算、搜索索引编制或缓存失效可能会增加延迟时间并导致共享行上的锁争用，从而降低用户体验。

脏标记模式可将数据修改与异步处理分离。当事务修改表时，它会在同一事务中将轻量级“脏位”消息写入队列。后台工作器随后会使用该消息并异步执行开销较大的处理。

例如，当客户更改个人资料或设置时：

GoogleSQL

-- Inside user profile update transaction:
-- 1. Update the primary entity table
UPDATE UserProfiles
SET FullName = 'Jane Doe', UpdatedAt = CURRENT_TIMESTAMP()
WHERE UserId = 456;

-- 2. Send lightweight dirty flag message to the queue.
-- It is recommended to interleave the queue in the primary UserProfiles
-- table for better transaction performance.
INSERT INTO UserDirtyQueue (UserId, TaskType, CommitTimestamp, Payload)
VALUES (456, 'reindex-user-profile', CURRENT_TIMESTAMP(), b'');

PostgreSQL

-- Inside user profile update transaction:
-- 1. Update the primary entity table
UPDATE userprofiles
SET fullname = 'Jane Doe', updatedat = CURRENT_TIMESTAMP
WHERE userid = 456;

-- 2. Send lightweight dirty flag message to the queue
INSERT INTO userdirtyqueue (userid, tasktype, committimestamp, payload)
VALUES (456, 'reindex-user-profile', CURRENT_TIMESTAMP, CAST('' AS bytea));

UserDirtyQueue 的后台接收器接收 UserId，在用户关键路径之外读取新的个人资料行，并重新计算搜索索引或更新外部缓存。如果短时间内向队列发送了针对同一用户的多个更新，也可以将批处理应用于此模式。在这种情况下，接收器 TVF 可以指定大于 1 的 max_batch_size 来接收同一批次中的多条消息。

监控 worker 健康状况并检测超时

您可以使用预定队列消息为工作器节点或微服务实例的集群构建容错心跳和健康状况监控系统。

如需实现健康检查，请执行以下操作：

在启动时注册工作器：当工作器初始化时，它会将一条心跳消息插入到健康检查队列中，并将未来的 DeliverTime 设置为故障截止时间（例如 60 秒）。
发送周期性心跳：在正常运行期间，工作器会定期（例如每 10 秒）通过将 DeliverTime 推进到未来 60 秒来刷新其心跳消息。
检测故障：如果工作器崩溃或失去网络连接，心跳刷新会停止。60 秒后，传送时间戳成熟 (DeliverTime <= CURRENT_TIMESTAMP())，Spanner 会将消息传送给提醒接收者，后者会启动故障切换或任务重新分配。

重要提示：Spanner 队列不支持 UPDATE DML 语句。因此，如需刷新心跳时间戳，您必须在单个事务中删除现有消息并插入包含新 DeliverTime 的替换消息，或者应用客户端库 Ack 和 Send 突变。确保队列的主键仅为 WorkerId（而非 (WorkerId, MessageId)），以便在任何给定时间每个工作器仅存在一条心跳消息。

GoogleSQL

-- Inside the worker heartbeat transaction (executed every 10 seconds):
-- 1. Acknowledge the existing heartbeat message
DELETE FROM WorkerHealthQueue
WHERE WorkerId = 'worker-node-42' ASSERT_ROWS_MODIFIED 1;

-- 2. Send replacement heartbeat with refreshed 60-second deadline
INSERT INTO WorkerHealthQueue (WorkerId, DeliverTime, Payload)
VALUES (
  'worker-node-42',
  TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 60 SECOND),
  b'{"status": "healthy", "active_jobs": 3}'
);

PostgreSQL

-- Inside the worker heartbeat transaction (executed every 10 seconds):
-- 1. Acknowledge the existing heartbeat message
DELETE FROM workerhealthqueue
WHERE workerid = 'worker-node-42' ASSERT_ROWS_MODIFIED 1;

-- 2. Send replacement heartbeat with refreshed 60-second deadline
INSERT INTO workerhealthqueue (workerid, deliver_time, payload)
VALUES (
  'worker-node-42',
  CURRENT_TIMESTAMP + INTERVAL '60 SECOND',
  CAST('{"status": "healthy", "active_jobs": 3}' AS bytea)
);

使用客户端库突变时，请确保突变切片中的 Ack 位于 Send 之前，如以下 Go 示例所示：

// Worker heartbeat loop
func sendHeartbeat(ctx context.Context, client *spanner.Client, workerID string) error {
    newDeadline := time.Now().Add(60 * time.Second)
    _, err := client.Apply(ctx, []*spanner.Mutation{
        spanner.Ack("WorkerHealthQueue", spanner.Key{workerID}),
        spanner.Send(
            "WorkerHealthQueue",
            spanner.Key{workerID},
            []byte(`{"status":"healthy"}`),
            spanner.WithDeliveryTime(newDeadline),
        ),
    })
    return err
}

当工作器正常关闭时，它会明确删除自己的心跳消息，以免触发错误提醒：

DELETE FROM WorkerHealthQueue WHERE WorkerId = 'worker-node-42' ASSERT_ROWS_MODIFIED 1;

在单个队列中处理多种任务类型

Spanner 实例对队列总数有限制。为每个小型异步操作创建单独的队列很快就会达到此限制，并且需要管理许多并发的接收器查询。

如需整合操作，请将不同的任务类型合并到单个队列中，该队列称为多态队列。创建多态队列有两种策略。

策略 1：在主键中添加类型列

GoogleSQL

CREATE QUEUE ApplicationTasks (
  TaskType   STRING(50) NOT NULL,
  TaskId     STRING(36) NOT NULL,
  Payload    BYTES(MAX) NOT NULL,
) PRIMARY KEY (TaskType, TaskId);

-- Enqueue an email task
INSERT INTO ApplicationTasks (TaskType, TaskId, Payload)
VALUES ('SEND_EMAIL', 'task-uuid-1', b'{"to": "user@example.com", "template": "welcome"}');

-- Enqueue an image thumbnail task
INSERT INTO ApplicationTasks (TaskType, TaskId, Payload)
VALUES ('GENERATE_THUMBNAIL', 'task-uuid-2', b'{"image_id": "img-789", "size": "small"}');

PostgreSQL

CREATE QUEUE applicationtasks (
  tasktype   varchar(50) NOT NULL,
  taskid     varchar(36) NOT NULL,
  payload    bytea NOT NULL,
  PRIMARY KEY (tasktype, taskid)
);

-- Enqueue an email task
INSERT INTO applicationtasks (tasktype, taskid, payload)
VALUES ('SEND_EMAIL', 'task-uuid-1', CAST('{"to": "user@example.com", "template": "welcome"}' AS bytea));

-- Enqueue an image thumbnail task
INSERT INTO applicationtasks (tasktype, taskid, payload)
VALUES ('GENERATE_THUMBNAIL', 'task-uuid-2', CAST('{"image_id": "img-789", "size": "small"}' AS bytea));

接收方检查 TaskType 并将载荷分派给相应的处理程序。

策略 2：多态载荷结构

或者，使用包含操作或类型判别器字段的 JSON 载荷：

{
  "action": "SYNC_INVENTORY",
  "data": { "item_id": 987, "delta": -1 }
}

实现自定义重试延迟

Spanner 队列会自动使用内置的指数退避算法重试失败或未确认的消息。不过，在以下情况下，如果消息因已知原因而失败，且持续时间已知，或者存在外部速率限制（例如指定了 Retry-After 标头的 HTTP 429 响应），依赖自动退避可能会导致过早的重试尝试，从而浪费 CPU 资源。

如需实现自定义重试延迟，请执行以下操作：

捕获消息处理器中的特定暂时性故障。
确认当前消息以满足当前传送尝试。
在同一交易中，发送一条替换消息，其中显式将 DeliverTime 设置为所选的未来重试时间（在以下示例中，为 5 分钟后）。

GoogleSQL

-- Inside failure-handling transaction:
-- 1. Acknowledge the failed message
DELETE FROM OutboundNotificationQueue
WHERE NotificationId = 'notif-555' ASSERT_ROWS_MODIFIED 1;

-- 2. Reschedule delivery 5 minutes in the future
INSERT INTO OutboundNotificationQueue (NotificationId, DeliverTime, Payload)
VALUES (
  'notif-555',
  TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 5 MINUTE),
  b'{"recipient": "user@example.com", "retry_count": 2}'
);

PostgreSQL

-- Inside failure-handling transaction:
-- 1. Acknowledge the failed message
DELETE FROM outboundnotificationqueue
WHERE notificationid = 'notif-555' ASSERT_ROWS_MODIFIED 1;

-- 2. Reschedule delivery 5 minutes in the future
INSERT INTO outboundnotificationqueue (notificationid, deliver_time, payload)
VALUES (
  'notif-555',
  CURRENT_TIMESTAMP + INTERVAL '5 MINUTE',
  CAST('{"recipient": "user@example.com", "retry_count": 2}' AS bytea)
);

此方法可让您的应用精确管理退避时间表，并避免在下游恢复期间使外部 API 饱和。

Spanner 队列场景和示例 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

“正好一次”处理和“最多一次”确认

在交易提交后执行工作

GoogleSQL

PostgreSQL

GoogleSQL

PostgreSQL

处理长时间运行的工作

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

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

GoogleSQL

PostgreSQL

处理大型消息载荷

GoogleSQL

PostgreSQL

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

定期执行操作

GoogleSQL

PostgreSQL

Java

Go

Python

Node.js

使用时间批处理功能批量处理消息

GoogleSQL

PostgreSQL

GoogleSQL

PostgreSQL

将数据修改与处理分离（脏标记模式）

GoogleSQL

PostgreSQL

监控 worker 健康状况并检测超时

GoogleSQL

PostgreSQL

在单个队列中处理多种任务类型

策略 1：在主键中添加类型列

GoogleSQL

PostgreSQL

策略 2：多态载荷结构

实现自定义重试延迟

GoogleSQL

PostgreSQL

后续步骤

Spanner 队列场景和示例