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

Question

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

c#.netentity-frameworkasynchronousasync-await

3

我想做的是从一张大约有四百万行的表中获取数据，将其与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; }
}

我有一个包含大约4百万行记录的表格，查询肯定需要一些时间。因此，我希望能够异步地进行查询。

我已经尝试了多种方法。首先是将查询分批，并在其上执行并行循环，但这会引发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异步操作来完成这个任务的方法？

- Ekenstein

indexModels 的大小是多少？ - i3arnon

有些索引模型设置大约20个属性，而有些只有5个左右。 - Ekenstein

多达约400万个索引模型。 - Ekenstein

为什么要使用ORM来爬取数据库，当ElasticSearch已经通过JDBC连接支持了这一点呢？你不会从使用ORM中获得任何好处（因为没有对象参与），但是你确实会增加显着的开销，因为需要通过一个中介来传递数据。 - Panagiotis Kanavos

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Yuval Itzchakov · Accepted Answer

使用自然的async API的优点是无需使用线程即可使用它们。因为一直到WinAPI级别，没有线程。

您可以创建一个方法，该方法需要一个IEnumerable<IndexModel>并使用ElasticSearch的异步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);
}

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