如何在.NET/C#的Cosmos DB事务批处理中执行查询+插入操作？

首先明确：Cosmos DB 的事务批处理（Transactional Batch）不支持包含查询/读操作，它仅针对同分区键的批量写操作（创建、替换、删除等）提供原子性保障。所以无法直接将查询步骤嵌入批处理中，需要通过其他方式实现你要的原子性读-计算-写逻辑，同时避免并发场景下的计数重复问题。

可行解决方案：乐观并发控制 + 事务批处理 #

针对你的计数日志场景，推荐维护一个单独的计数文档（而非每次统计Products数量），结合乐观并发控制和事务批处理，实现原子性的计数更新与日志插入，从根源避免并发冲突。

核心思路 #

1. 用一个专门的ProductCounter文档维护当前产品计数，而非每次查询Products集合统计数量
2. 读取计数文档时获取其ETag（版本标识），用于乐观并发校验
3. 构建事务批处理，包含两个原子操作：
   • 替换计数文档（用ETag确保读取后无其他修改）
   • 插入带新计数的日志文档
4. 若遇到并发冲突（ETag不匹配），自动重试直到成功

C# 代码示例 #

首先定义实体类：

// 产品计数维护文档
public class ProductCounter
{
    public string Id { get; set; } = "ProductCounter";
    public int Count { get; set; }
    // 需与Products、日志文档使用相同的分区键
    public string PartitionKey { get; set; } = "ProductPartition";
}

// 产品计数日志文档
public class ProductCountLog
{
    public string Id { get; set; } = Guid.NewGuid().ToString();
    public int Count { get; set; }
    public string PartitionKey { get; set; } = "ProductPartition";
    public DateTime CreatedAt { get; set; } = DateTime.UtcNow;
}

实现业务逻辑：

using Microsoft.Azure.Cosmos;

public async Task ProcessProductCountLogAsync(CosmosContainer container)
{
    var partitionKeyValue = "ProductPartition";
    ItemResponse<ProductCounter> counterResponse = null;

    // 循环处理并发冲突
    do
    {
        // 1. 读取当前计数文档
        counterResponse = await container.ReadItemAsync<ProductCounter>(
            id: "ProductCounter",
            partitionKey: new PartitionKey(partitionKeyValue));
        
        var currentCounter = counterResponse.Resource;
        int newCount = currentCounter.Count + 1;

        // 2. 构建事务批处理：原子更新计数 + 插入日志
        var batch = container.CreateTransactionalBatch(new PartitionKey(partitionKeyValue))
            // 替换计数文档，用原ETag做乐观并发校验
            .ReplaceItemAsync(
                id: currentCounter.Id,
                item: currentCounter with { Count = newCount },
                requestOptions: new ItemRequestOptions { IfMatchEtag = counterResponse.ETag })
            // 插入新的计数日志
            .CreateItemAsync(new ProductCountLog { Count = newCount });

        // 3. 执行批处理
        var batchResult = await batch.ExecuteAsync();

        // 批处理成功则退出循环
        if (batchResult.IsSuccessStatusCode)
        {
            break;
        }

        // 仅当遇到并发冲突（412前置条件失败）时重试
    } while (counterResponse.StatusCode == System.Net.HttpStatusCode.PreconditionFailed);
}

方案优势 #

• 原子性保障：事务批处理确保计数更新和日志插入要么同时成功，要么同时失败
• 并发安全：乐观并发控制（ETag）避免多个实例同时修改计数导致的重复日志
• 无服务器端代码：无需维护JavaScript存储过程，全部用C#客户端代码实现

关于原存储过程的替代说明 #

如果你的原始逻辑依赖统计Products集合的真实数量（而非维护独立计数文档），那存储过程依然是最直接的原子性实现方式——因为存储过程在Cosmos DB服务器端执行，可直接嵌入查询+写操作，天然保证原子性。但如果想替换为客户端事务批处理，就需要将“统计数量”改为“维护独立计数文档”的模式，否则无法避免并发场景下的计数重复问题。

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