基于GSI哈希键批量查询DynamoDB的多请求优化问询

问题分析 #

你的代码通过循环每个ProductId单独发起GSI查询，请求次数等于输入列表长度，会带来较高的网络开销和延迟。以下是两种可行的优化方案：

方案一：使用PartiQL批量查询（推荐） #

DynamoDB的PartiQL支持通过IN条件一次性查询多个GSI哈希键值，只需少量请求（按100个ID分批）即可获取所有匹配的ProductId，大幅减少请求次数。注意：IN条件最多支持100个值，若输入列表长度超过100，需要分批处理。

代码示例 #

using Amazon.DynamoDBv2;
using Amazon.DynamoDBv2.Model;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

public async Task<IEnumerable<string>> GetProductListWithMissingTransportMissionAsync(IEnumerable<string> listProductId, CancellationToken cancellationToken = default)
{
    var productIds = listProductId.ToList();
    var availableProductIds = new HashSet<string>();

    // 按DynamoDB限制分批，每批最多100个ID
    for (int i = 0; i < productIds.Count; i += 100)
    {
        var batch = productIds.Skip(i).Take(100);
        // 构造PartiQL查询语句
        var query = $"SELECT ProductId FROM \"{_tableName}\".\"TransportMissionProductIdIndex\" WHERE ProductId IN [{string.Join(",", batch.Select(id => $"'{id}'"))}]";
        
        var request = new ExecuteStatementRequest
        {
            Statement = query
        };

        var response = await _dynamoDbClient.ExecuteStatementAsync(request, cancellationToken);
        // 提取返回的ProductId
        foreach (var item in response.Items)
        {
            if (item.TryGetValue("ProductId", out var value))
            {
                availableProductIds.Add(value.S);
            }
        }
    }

    // 返回未匹配的ProductId
    return productIds.Except(availableProductIds);
}

说明：

• 需要注入IAmazonDynamoDB客户端（需额外配置，而非仅依赖IDynamoDBContext）
• 使用HashSet存储已匹配的ID，提升后续差集计算的效率
• 自动处理超100个ID的分批逻辑，符合DynamoDB的API限制

方案二：并行异步查询 #

如果无法使用PartiQL，可以将循环查询改为并行执行，通过Task.WhenAll或Parallel.ForEachAsync并发处理多个请求，减少总耗时。注意需控制并行度，避免触发DynamoDB的并发请求限额。

代码示例 #

using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

public async Task<IEnumerable<string>> GetProductListWithMissingTransportMissionAsync(IEnumerable<string> listProductId, CancellationToken cancellationToken = default)
{
    var productIds = listProductId.ToList();
    var availableProductIds = new ConcurrentHashSet<string>();

    // 控制并行度，避免超出DynamoDB账户并发限制
    var parallelOptions = new ParallelOptions
    {
        MaxDegreeOfParallelism = 10,
        CancellationToken = cancellationToken
    };

    await Parallel.ForEachAsync(productIds, parallelOptions, async (productId, ct) =>
    {
        var items = await _context.QueryAsync<TransportMissionDb>(productId,
            new DynamoDBOperationConfig()
            {
                OverrideTableName = _tableName,
                IndexName = "TransportMissionProductIdIndex"
            }).GetRemainingAsync(ct);
        
        if (items.Any())
        {
            availableProductIds.Add(items.First().ProductId);
        }
    });

    return productIds.Except(availableProductIds);
}

说明：

• 使用ConcurrentHashSet保证多线程环境下的线程安全（需引入System.Collections.Concurrent命名空间）
• 通过MaxDegreeOfParallelism限制并发数，避免触发DynamoDB的限流机制
• 总请求次数仍等于列表长度，但并行执行能显著缩短整体耗时

方案对比 #

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