为两个Azure表存储实现发布/订阅总线及表数据同步需求问询

针对你这个Azure表存储双向同步+订阅跟踪的需求，我整理了几个实际项目中用过的可行方案，你可以根据自己的技术栈来选：

咱们要实现的是双向联动：

• 表1更新时，自动同步表2里已存在的对应实体（因为表2是子集，只同步它关心的那些）
• 表2新增实体时，通知表1“以后这个实体的更新要同步给我”，后续表1更新时就自动同步过去

这是最灵活的方案，适合有开发能力的场景，不用自己写轮询逻辑，靠触发器自动触发。

1. 表1更新 → 同步到表2 #

先建一个Azure Function，用Table Storage Trigger监听表1的插入/更新操作：

• 触发器会在表1有实体新增或修改时自动跑
• 函数里先检查表2有没有这个实体（用PartitionKey+RowKey匹配）
• 如果有，就把表1的更新字段同步过去；没有就跳过（毕竟表2是子集，只同步它已经有的）

给你个C#的示例代码：

public static async Task SyncTable1ToTable2(
    [TableTrigger("Table1")] MyBusinessEntity updatedTable1Entity,
    [Table("Table2")] CloudTable table2,
    [Table("TrackedEntities")] CloudTable trackedTable, // 后面要用到的跟踪表
    ILogger log)
{
    // 先查这个实体是否在跟踪清单里（表2新增时会加进来）
    var trackCheckOp = TableOperation.Retrieve<TrackedEntity>(updatedTable1Entity.PartitionKey, updatedTable1Entity.RowKey);
    var trackResult = await trackedTable.ExecuteAsync(trackCheckOp);

    if (trackResult.Result != null && ((TrackedEntity)trackResult.Result).IsActive)
    {
        // 检查表2中是否存在该实体
        var table2CheckOp = TableOperation.Retrieve<MyBusinessEntity>(updatedTable1Entity.PartitionKey, updatedTable1Entity.RowKey);
        var table2Result = await table2.ExecuteAsync(table2CheckOp);

        if (table2Result.Result != null)
        {
            var existingTable2Entity = (MyBusinessEntity)table2Result.Result;
            // 同步需要的字段，比如把表1的最新值覆盖过去
            existingTable2Entity.Status = updatedTable1Entity.Status;
            existingTable2Entity.LastUpdated = updatedTable1Entity.LastUpdated;

            var updateOp = TableOperation.Replace(existingTable2Entity);
            await table2.ExecuteAsync(updateOp);
            log.LogInformation($"Synced entity {updatedTable1Entity.RowKey} from Table1 to Table2");
        }
    }
}

2. 表2新增实体 → 让表1开始跟踪 #

再建另一个Azure Function，用Table Storage Trigger只监听表2的新增操作（更新不用管）：

• 当表2新增实体时，把这个实体的标识（PartitionKey+RowKey）存入一个专门的「跟踪清单表」（比如叫TrackedEntities）
• 这样表1的同步函数就能通过这个清单判断要不要同步某个实体的更新

示例代码：

public static async Task AddEntityToTrackingList(
    [TableTrigger("Table2", Filter = "InsertionTime ge datetime'1970-01-01'")] MyBusinessEntity newTable2Entity,
    [Table("TrackedEntities")] CloudTable trackedTable,
    ILogger log)
{
    // 把新增实体的标识存入跟踪表，标记为需要跟踪
    var trackedEntity = new TrackedEntity
    {
        PartitionKey = newTable2Entity.PartitionKey,
        RowKey = newTable2Entity.RowKey,
        IsActive = true,
        AddedTime = DateTime.UtcNow
    };

    var insertOp = TableOperation.InsertOrReplace(trackedEntity);
    await trackedTable.ExecuteAsync(insertOp);
    log.LogInformation($"Added entity {newTable2Entity.RowKey} to tracking list");
}

如果不想写代码，用Logic Apps的可视化拖拽就能搞定，适合非开发人员或者快速搭建场景：

• 第一个Logic App：触发条件选「当Azure表存储中添加或修改实体时」（指向表1），然后加动作「获取Azure表存储中的实体」（指向表2），判断实体存在就执行「更新Azure表存储中的实体」
• 第二个Logic App：触发条件选「当Azure表存储中添加实体时」（指向表2），然后加动作「插入或替换Azure表存储中的实体」到「TrackedEntities」跟踪表

• 幂等性：表存储触发器可能会重复触发，所以同步逻辑要保证重复执行不会出问题（比如用InsertOrReplace代替Insert，或者给实体加个版本号字段，只有新版本才同步）
• 性能优化：如果表1更新特别频繁，可以设置触发器的批量大小（一次处理多个实体）；跟踪表按PartitionKey分区，提升查询速度
• 错误处理：在函数或Logic Apps里加重试策略（比如失败后重试3次），同时记录错误日志，方便排查同步失败的情况

内容的提问来源于stack exchange，提问作者A03