如何用Node.js的@google-cloud/bigtable安全并发更新Bigtable非整数字符串列？

针对你遇到的Bigtable并发更新JSON字符串的问题，以下是几种无需大规模重构数据库的可行方案：

1. 条件更新（Check-and-Set, CAS）实现原子修改 #

利用Bigtable的checkAndMutateRow原子操作，先读取当前JSON值，修改后仅当单元格值未被其他请求改动时才写入，从根源避免竞态条件。适合需要修补或增量更新JSON结构的场景。

代码示例 #

const { Bigtable } = require('@google-cloud/bigtable');
const bigtable = new Bigtable();

async function atomicJsonUpdate(tableId, rowKey, columnFamily, column, updateLogic) {
  const table = bigtable.table(tableId);
  const row = table.row(rowKey);
  const maxRetries = 5;
  let retryCount = 0;

  while (retryCount < maxRetries) {
    try {
      // 读取当前单元格的JSON值
      const [rowData] = await row.get({ columns: [`${columnFamily}:${column}`] });
      const currentCell = rowData[`${columnFamily}:${column}`]?.[0];
      const currentValue = currentCell ? currentCell.value : '';
      const currentJson = currentValue ? JSON.parse(currentValue) : {};

      // 应用自定义更新逻辑
      const updatedJson = updateLogic(currentJson);
      const updatedValue = JSON.stringify(updatedJson);

      // 条件更新：仅当当前值与读取值一致时执行写入
      const [updateSuccess] = await row.checkAndMutateRow(`${columnFamily}:${column}`)
        .equal(currentValue)
        .then(row.mutate({
          method: 'insert',
          column: `${columnFamily}:${column}`,
          value: updatedValue
        }))
        .run();

      if (updateSuccess) {
        return updatedJson;
      }
      // 冲突则重试，加入指数退避避免频繁请求
      retryCount++;
      await new Promise(resolve => setTimeout(resolve, 100 * Math.pow(2, retryCount)));
    } catch (err) {
      console.error('更新失败:', err);
      retryCount++;
      await new Promise(resolve => setTimeout(resolve, 100 * Math.pow(2, retryCount)));
    }
  }
  throw new Error('超出最大重试次数，更新失败');
}

// 使用示例：给JSON添加新字段或修改现有字段
atomicJsonUpdate('my-table', 'user-123', 'user_data', 'profile', (current) => {
  current.last_login = new Date().toISOString();
  current.tags = [...(current.tags || []), 'updated'];
  return current;
});

2. 原子字符串追加（针对追加场景） #

如果你的需求以追加内容为主（比如给JSON数组添加元素），可以利用Bigtable的append原子操作直接追加字符串片段，跳过读取-修改-写入的流程，彻底避免竞态。

代码示例 #

async function appendJsonElement(tableId, rowKey, columnFamily, column, newElement) {
  const table = bigtable.table(tableId);
  const row = table.row(rowKey);

  // 将元素转为JSON字符串，前置逗号处理初始空值情况
  const elementStr = `,${JSON.stringify(newElement)}`;

  // 原子追加操作
  await row.mutate({
    method: 'append',
    column: `${columnFamily}:${column}`,
    value: elementStr
  });

  // 读取时整理为标准JSON数组
  const [rowData] = await row.get({ columns: [`${columnFamily}:${column}`] });
  const rawValue = rowData[`${columnFamily}:${column}`]?.[0]?.value || '';
  const cleanedValue = rawValue.startsWith(',') ? rawValue.slice(1) : rawValue;
  return cleanedValue ? JSON.parse(`[${cleanedValue}]`) : [];
}

// 使用示例：追加一条操作日志到JSON数组
appendJsonElement('my-table', 'user-123', 'user_data', 'activity_logs', {
  action: 'login',
  timestamp: new Date().toISOString()
});

3. 最小化拆分列族/列 #

如果JSON中有特定字段频繁更新，可以将这些字段单独拆分为独立列，保留原有JSON列不变。对拆分出的列使用原子操作（如increment或append），读取时再合并原有JSON和独立列的数据，兼顾兼容性和并发安全性。

比如原JSON有count（需增量）和logs（需追加），可以把count设为单独列用increment，logs用单独列做append，原有JSON列保留其他不常更新的字段。

内容的提问来源于stack exchange，提问作者Davi E. Silva