如何利用“Outbox模式”来解决Node.js中的双写问题

请求处理程序

整个流程就是从这里开始的。请求处理程序会接收HTTP POST请求，将订单信息插入数据库，并在outbox表中添加相应的记录，最后通过一次原子性事务完成所有操作。关键在于：只有当这两项操作都成功时，整个流程才会完成。

const client = await pool.connect();
try {
  await client.query('BEGIN');

  // 插入订单记录
  const { rows } = await client.query(
    'INSERT INTO orders (customer_id, amount_cents) VALUES ($1, $2) RETURNING *',
    [customerId, amountCents]
  );
  const order = rows[0];

  // 在同一事务中插入outbox记录
  await client.query(
    `INSERT INTO outbox (event_type, payload)
     VALUES ($1, $2)`,
    ['order.created', JSON.stringify({ orderId: order.id, customer_id: order.customer_id, amount_cents: order.amount_cents, created_at: order.created_at })],
  );

  await client.query('COMMIT');
  res.status(201).json(order);
} catch (err) {
  await client.query('ROLLBACK');
  next(err);
} finally {
  client.release();
}

处理程序会从请求体中获取`customerId`和`amountCents`，然后使用`BEGIN`语句启动一个事务，并将订单信息插入数据库。随后，它还会在数据库的“待发送”表中添加一条记录，将这些订单数据作为该记录的内容。

所有操作都会以原子性方式完成。如果其中任何一步失败，所有操作都会被回滚，客户端也会收到错误提示。如果在提交操作和返回响应之间系统发生崩溃，客户端虽然不会收到201状态码，但订单信息以及“待发送”表中的记录仍然会安全地保存在数据库中，后续的Relay进程会自动处理这些数据。需要注意的是，处理程序本身并不直接调用SQS服务，这一任务是由Relay进程来完成的。

Relay进程

Relay进程是一个独立的进程，它每秒钟会检查一次“待发送”表，并将其中尚未被处理的记录发送到SQS服务器。该进程与HTTP服务器是独立运行的，两者之间也没有任何共享状态。

async function relay() {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    const { rows } = await client.query(`
      SELECT *
      FROM outbox
      WHERE status = 'pending'
      ORDER BY created_at
      LIMIT 10
      FOR UPDATE SKIP LOCKED -- 防止多个Relay进程同时处理同一批记录
    `);

    for (const row of rows) {
      await sqsClient.send(new SendMessageCommand({
        QueueUrl: QUEUE_URL,
        MessageBody: JSON.stringify(row.payload),
        MessageAttributes: {
          EventType: { DataType: 'String', StringValue: row.event_type },
        },
      });

      await client.query(
        `UPDATE outbox SET status = 'sent', sent_at = now() WHERE id = $1`,
        [row.id],
      );
    }

    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    console.error('Relay error:', err.message);
  } finally {
    client.release();
  }
}

setInterval(relay, 1000);

FOR UPDATE SKIP LOCKED这一语句是确保多个Relay进程能够安全运行的关键：当某个Relay进程开始处理一批记录时，它会将这些记录锁定起来；其他Relay进程如果尝试访问这些被锁定的记录，就会跳过它们，继续处理下一批记录。因此，永远不会有两个Relay进程同时发送相同的内容。

只有当`sqsClient.send`方法执行完成之后，Relay进程才会将对应的记录状态标记为“已发送”。如果在向SQS服务器发送数据后但尚未更新数据库记录之前，Relay进程发生崩溃，那么这些记录的状态仍然会保持为“待发送”，Relay进程会在下一次运行时重新尝试发送这些数据。

需要注意的是，`UPDATE`操作是在与`SELECT FOR UPDATE`相同的事务中完成的。因此，如果Relay进程在处理一批记录的过程中发生崩溃，那么这一批记录都会被回滚，所有记录都需要重新处理，包括那些已经成功发送到SQS服务器的记录。

“至少一次交付”的保证是针对整批记录而言的，而不是针对单条记录。关于这个问题，你可以参考《超越快乐路径工程：网络层面的设计考虑》这篇文章：当某个响应丢失时，调用方无法确定操作是否成功，因此会重新尝试发送请求，而接收方可能会收到同一条请求两次。这就是为什么在消费者端确保操作的幂等性如此重要。

消费者服务

消费者服务是一个完全独立的服务。它对订单服务所使用的 PostgreSQL 数据库一无所知。它的唯一输入是 SQS 队列，而唯一的输出则是 DynamoDB 中的 fulfillments 表。这种设计的关键在于：通过队列将这两项服务解耦开来，使得每一项服务都拥有自己独立的数据存储系统。

正如我们前面所提到的，由于 SQS 至少会交付一次消息（也就是说，同一条消息可能会被多次传递），因此消费者服务必须具备幂等性。在调用 PutItem 方法时，会使用一个 ConditionExpression 来确保：如果该订单已经有了对应的完成记录，那么这次写入操作就会被视为无效操作，从而保证重复传递的消息能够被正确处理。

async function consume() {
  const { Messages } = await sqsClient.send(new ReceiveMessageCommand({
    QueueUrl:              QUEUE_URL,
    WaitTimeSeconds:       20,   // 长轮询：最多等待 20 秒以接收消息
    MaxNumberOfMessages:   10,
    MessageAttributeNames: ['All'],
  }));

  for (const msg of Messages ?? []) {
    const event = JSON.parse(msg.Body);

    try {
      await dynamoClient.send(new PutItemCommand({
        TableName: 'fulfillments',
        Item: {
          orderId:     { S: event.orderId },
          customerId:  { S: event.customerId },
          amountCents: { N: String(event.amountCents) },
          status:      { S: 'received' },
         createdAt:   { S: new Date().toISOString() },
        },
        ConditionExpression: 'attribute_not_exists(orderId)', // 幂等性检查
      }));
    } catch (err) {
      if (err.name !== 'ConditionalCheckFailedException') throw err;
      // 如果已经处理过这条消息，就可以继续执行后续操作。
    }

    // 只有在写入操作成功之后，才会删除该消息。
    await sqsClient.send(new DeleteMessageCommand({
      QueueUrl:      QUEUE_URL,
      ReceiptHandle: msg.ReceiptHandle,
    }));
  }
}

ConditionExpression: 'attribute_not_exists(orderId)' 这条条件语句会让 DynamoDB 在遇到已经存在具有相同 orderId 的记录时，拒绝执行写入操作。当这种情况发生时，DynamoDB 会抛出 ConditionalCheckFailedException 异常。消费者服务会捕获这种特定的异常并忽略它，然后从队列中删除该消息并继续执行后续任务。而对于其他类型的异常，系统会重新抛出错误，导致消息仍然留在队列中，等待下一次尝试被处理。

DeleteMessage 操作是在 DynamoDB 完成写入操作之后才执行的，而不是在之前。如果在写入操作和删除操作之间进程发生了崩溃，SQS 会再次传递该消息，此时条件检查机制会确保消息能够被正确处理。如果进程在写入操作之前就崩溃了，那么消息就会留在队列中，等待下一次交付时再被处理。

整体运行流程

当 floci 服务已经启动并且所有资源都已经配置完毕之后，打开三个终端窗口，分别运行以下进程：

node src/server.js    # 这个进程负责提供订单相关的 API，运行在 3000 端口
node src/relay.js     # 负责中转消息的进程
node src/consumer.js  # 负责处理完成通知的进程

现在，请下订单：

curl -X POST localhost:3000/orders \
  -H 'Content-Type: application/json' \
  -d '{"customerId":"c1","amountCents":4999}'

您应该会收到一个状态码为201的响应，其中包含新的订单记录：

{"id":"1768d35b-083d-45f1-adb5-4063d8d7fcab","customer_id":"c1","amount_cents":4999,"created_at":"2026-07-30T20:27:10.628Z"}

在一秒钟之内，中继系统会读取这些订单信息并将其发布到SQS队列中。消费者接收到这些消息后，会将相应的处理结果记录到DynamoDB数据库中。该代码库中还包含一个用于验证这一流程的脚本：

npm run check

您应该会看到一条包含您刚刚下订单的orderId的处理记录：

{
  orderId: 'c335640e-bc4a-47e4-afed-484c95fbd6d3',
  customer_id: 'c1',
  amountCents: '4999',
  status: 'received',
  creado_at: '2026-07-30T19:02:54.929Z'
}

投入生产环境

由于本地开发环境使用floci来模拟AWS的功能，因此切换到真正的AWS环境时完全不需要修改任何代码。AWS SDK会从环境变量AWS_ENDPOINT_URL中获取端点信息；在生产环境中，只需不设置该变量即可，SDK会使用标准环境变量中的凭证和区域信息来与真实的AWS系统进行交互（这些变量包括AWS_REGION、AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY，或者在运行在EC2或ECS上时使用的IAM角色）。

由于使用了FOR UPDATE SKIP LOCKED机制，同时运行多个中继实例也是完全安全的。您可以水平扩展这些中继节点，每个节点都会处理不同的一组数据，从而避免消息重复。

在将系统投入生产环境之前，还需要考虑如何处理中继系统中可能出现的永久性故障。目前，中继系统只区分pending和sent两种状态，因此应该添加failed状态以及重试机制：当某条记录经过N次尝试仍然无法成功处理时，应将其标记为failed并停止重试。同时，还需要在orders SQS队列上配置一个死信队列，这样那些经过最大次数重试后仍无法被处理的消息就会被保存到这个队列中，以便后续进行检查。

对于那些对polling延迟要求较高的高吞吐量系统来说，变更数据捕获技术是一种常见的替代方案。像Debezium这样的工具可以直接从PostgreSQL的写前日志中读取数据变化，并将其实时发布到Kafka或SQS队列中，而不会产生任何polling延迟。在这种情况下，只需要替换原有的中继系统即可，其他组件都可以保持不变。

不过，与传统的polling方式相比，采用变更数据捕获技术需要投入更多的运营资源，因此对于大多数系统来说，polling方式仍然是更合适的起点。

结论