IDのリストがあり、各IDでいくつかのストアドプロシージャを実行する必要があります。
標準のforeachループを使用している場合は問題なく機能しますが、レコードが多い場合はかなり遅くなります。
EFで動作するようにコードを変換したかったのですが、「基になるプロバイダーがOpenで失敗しました」という例外が発生します。
Parallel.ForEach内でこのコードを使用しています：
using (XmlEntities osContext = new XmlEntities())
{
//The code
}
しかし、それでも例外がスローされます。
Parallel with EFをどのように使用できますか？実行しているすべてのプロシージャに対して新しいコンテキストを作成する必要がありますか？私は約10の手順を持っているので、それぞれに1つずつ、10のコンテキストを作成するのは非常に悪いと思います。
Entity Framework が使用している基になるデータベース接続は、スレッドセーフではありません。実行する別のスレッドで操作ごとに新しいコンテキストを作成する必要があります。
操作を並列化する方法についての懸念は有効です。多くのコンテキストは、開いたり閉じたりするのにコストがかかります。
代わりに、コードの並列化についての考え方を逆にしたい場合があります。多数のアイテムをループしてから、アイテムごとにストアド プロシージャを順番に呼び出しているようです。
可能であれば、プロシージャごとに新しいTask<TResult>(またはTask、結果が必要ない場合は ) を作成し、その中で1 つのコンテキストを開き、すべての項目をループしてから、ストアド プロシージャを実行します。この方法では、並行して実行しているストアド プロシージャの数と同じ数のコンテキストしかありません。Task<TResult>
MyDbContextと の2 つのストアド プロシージャを持つ がDoSomething1ありDoSomething2、どちらもクラスのインスタンスを取ると仮定しますMyItem。
上記を実装すると、次のようになります。
// You'd probably want to materialize this into an IList<T> to avoid
// warnings about multiple iterations of an IEnumerable<T>.
// You definitely *don't* want this to be an IQueryable<T>
// returned from a context.
IEnumerable<MyItem> items = ...;
// The first stored procedure is called here.
Task t1 = Task.Run(() => {
// Create the context.
using (var ctx = new MyDbContext())
// Cycle through each item.
foreach (MyItem item in items)
{
// Call the first stored procedure.
// You'd of course, have to do something with item here.
ctx.DoSomething1(item);
}
});
// The second stored procedure is called here.
Task t2 = Task.Run(() => {
// Create the context.
using (var ctx = new MyDbContext())
// Cycle through each item.
foreach (MyItem item in items)
{
// Call the first stored procedure.
// You'd of course, have to do something with item here.
ctx.DoSomething2(item);
}
});
// Do something when both of the tasks are done.
ストアド プロシージャを並列に実行できない場合(それぞれが特定の順序で実行されることに依存しています)、操作を並列化することはできますが、それはもう少し複雑です。
アイテム全体にカスタム パーティションを作成する方法を検討します ( classの静的Createメソッドを使用)。これにより、実装を取得する手段が得られます (注意してください。これはそうではないため、それを超えることはできません)。PartitionerIEnumerator<T> IEnumerable<T>foreach
返されたインスタンスごとIEnumerator<T>に新しいインスタンスを作成しTask<TResult>(結果が必要な場合)、本体でコンテキストを作成し、ストアド プロシージャを順番に呼び出して、Task<TResult>によって返されたアイテムを循環します。IEnumerator<T>
それは次のようになります。
// Get the partitioner.
OrdinalPartitioner<MyItem> partitioner = Partitioner.Create(items);
// Get the partitions.
// You'll have to set the parameter for the number of partitions here.
// See the link for creating custom partitions for more
// creation strategies.
IList<IEnumerator<MyItem>> paritions = partitioner.GetPartitions(
Environment.ProcessorCount);
// Create a task for each partition.
Task[] tasks = partitions.Select(p => Task.Run(() => {
// Create the context.
using (var ctx = new MyDbContext())
// Remember, the IEnumerator<T> implementation
// might implement IDisposable.
using (p)
// While there are items in p.
while (p.MoveNext())
{
// Get the current item.
MyItem current = p.Current;
// Call the stored procedures. Process the item
ctx.DoSomething1(current);
ctx.DoSomething2(current);
}
})).
// ToArray is needed (or something to materialize the list) to
// avoid deferred execution.
ToArray();
これは私が使用しているもので、うまく機能します。さらに、エラー例外の処理をサポートし、追跡をはるかに容易にするデバッグモードを備えています
public static ConcurrentQueue<Exception> Parallel<T>(this IEnumerable<T> items, Action<T> action, int? parallelCount = null, bool debugMode = false)
{
var exceptions = new ConcurrentQueue<Exception>();
if (debugMode)
{
foreach (var item in items)
{
try
{
action(item);
}
// Store the exception and continue with the loop.
catch (Exception e)
{
exceptions.Enqueue(e);
}
}
}
else
{
var partitions = Partitioner.Create(items).GetPartitions(parallelCount ?? Environment.ProcessorCount).Select(partition => Task.Factory.StartNew(() =>
{
while (partition.MoveNext())
{
try
{
action(partition.Current);
}
// Store the exception and continue with the loop.
catch (Exception e)
{
exceptions.Enqueue(e);
}
}
}));
Task.WaitAll(partitions.ToArray());
}
return exceptions;
}
次のように使用します。db は元の DbContext であり、db.CreateInstance() は同じ接続文字列を使用して新しいインスタンスを作成します。
var batch = db.Set<SomeListToIterate>().ToList();
var exceptions = batch.Parallel((item) =>
{
using (var batchDb = db.CreateInstance())
{
var batchTime = batchDb.GetDBTime();
var someData = batchDb.Set<Permission>().Where(x=>x.ID = item.ID).ToList();
//do stuff to someData
item.WasMigrated = true; //note that this record is attached to db not batchDb and will only be saved when db.SaveChanges() is called
batchDb.SaveChanges();
}
});
if (exceptions.Count > 0)
{
logger.Error("ContactRecordMigration : Content: Error processing one or more records", new AggregateException(exceptions));
throw new AggregateException(exceptions); //optionally throw an exception
}
db.SaveChanges(); //save the item modifications
これが私の問題を解決するために私がしたことです。Parallel.ForEach 内で EF への独自の接続を使用して注文を取得しようとしていましたが、これは私が考えていたことですが、実際にはそうではなく、毎回同じ最後の注文をしていました。操作の最後に使用を削除して接続を閉じた後、正常に機能していました。
一言で言えば、using を利用する場合は、Parallel.Foreach 内の操作の最後に接続を閉じないでください。
これを交換しました
using ApplicationDbContext Context = CubiCommon.GetContext<ApplicationDbContext>("APIConnection");
これで
ApplicationDbContext Context = CubiCommon.GetContext<ApplicationDbContext>("APIConnection");
操作の最後にこれを追加しました
Context.Dispose();
Context = null;
多分これは同様の問題を解決することができます