使用 EF6 从数据库中选择大量异步数据

Question

我要做的是从大约有 400 万行的表中获取行，以针对 ElasticSearch 对其进行索引。

底层索引器将使用 IndexManyAsync 并批量处理提供给它的枚举。

类似：

public void IndexMany(IEnumerable<IIndexModel> indexModels) {
    var client = new ElasticClient(settings);
    var batches = indexModels.Batch(1000);
    var tasks = new List<Task>();
    Parallels.ForEach(partitions, partition =>
    {
        var task = client.IndexManyAsync(partition);
        tasks.Add(task);
    }

    Task.WaitAll(tasks.ToArray());
}

因此，考虑到这一点，我想使用 IndexModels 创建一个可枚举。

IndexModels 将获取一个实体并通过给定实体初始化各种属性。比如：

public class FooModel<T> : IIndexModel
{
    public FooModel(T entity) 
    {
        Name = entity.Name;
    }

    public string Name { get; set; }
}

我有一个包含约 40 万行的表，这显然需要一些时间来查询。所以我想做的是异步执行。

我已经尝试了各种方法来做到这一点。第一种方法是批量查询并对其进行并行处理。这给 ObjectContext 带来了各种并发问题。

public void IndexAllModels() {
    using (var db = new Db()) {
        var batchedEntities = db.BigTable.Select(p => p).Batch(1000);

        Parallels.ForEach(batchedEntities, currentBatch =>
        {
            var indexModels = new List<IIndexModel>();
            foreach (var entity in currentBatch) 
            {
                var indexModel = new FooModel<BigTable>(entity);
                indexModels.Add(indexModel);
            }

            IndexMany(indexModels);
        }
    }
}

我想知道是否有任何方法可以通过使用新的 EF6 异步操作来做到这一点？

Answer 1

自然使用async API 的好处是您不需要使用线程来使用它们。一直到 WinAPI 级别，There Is No Thread。

您可以创建一个采用 IEnumerable<IndexModel> 的方法并使用 ElasticSearchs 异步 API，如下所示：

public async Task IndexManyAsync(IEnumerable<IIndexModel> indexModels) 
{
    var client = new ElasticClient(settings);

    var taskBatches = indexModels.Batch(1000)
                                 .Select(partition =>
                                         client.IndexManyAsync(partition));

    await Task.WhenAll(taskBatches);
}

假设IndexManyAsync 为每个请求使用单独的DbContext，这应该可以工作。