c＃マルチスレッドデータベースの塗りつぶしテーブル

Question

私は小さなジレンマを持っています。私は、独自の接続文字列（同じか異なる可能性があります）を持つデータセット内のクエリのリストを調べ、すべてのDataTablesを埋め込み、すべてのクエリを埋め込んだ塗りつぶしデータセットを返す関数を持っています。

現在このプロセスは1つずつ実行されるため、クエリーの1つに10分かかり、残りの3つのクエリーにそれぞれ2分かかると、実行時間は16分になります。

この状況でマルチスレッドを使用できるかどうかは疑問でした。これは別のスレッドでFillTableをすべて呼び出す必要があり、実行時間を10分に短縮する必要があります。これらは、明示的にFill DataTable呼び出しのみです（更新または削除呼び出しはありません）。

これは私がこれまでに得たものである：

public void FillDataSet(ref DataSet Source) 
    { 
     foreach (var r in Source.Tables["queries"].Rows) 
     { 
      string query = r["QueryStatement"].ToString(); 
      string qSource = r["QuerySource"].ToString(); 
      string tableName = r["TableName"].ToString(); 

      DBConnection db = new DBConnection(); 
      var TempSource = Source; 
      taskConnection = Task.Factory.StartNew(() => callDB(db, query, tableName, ref TempSource)); 
      Source = TempSource; 
     } 
     Task.WaitAll(taskConnection); 
    } 

    private void callDB(DBConnection db, string query, string tableName, ref DataSet Source) 
    { 
     using (var sql = new SqlConnection(db.ConnectionString)) 
     { 
      sql.Open(); 
      using (var adp = new SqlDataAdapter(query, sql)) 
      { 
       adp.SelectCommand.CommandTimeout = 0; 
       adp.Fill(Source, tableName); 
      } 
     } 
    } 

ラムダ式は、（私はこれを変更することはできません）パラメータの参照を渡して好きではないので、私はTempSourceを作成する必要がありました。現在のところ、これは動作しません、私は間違っているか分からない。

Answer 1

ここには、使用できる基本的な定型文があります。 - 新しいスレッドを産卵のオーバーヘッドを作業量に比べて最小限であるので、私が代わりにスレッドプールのスレッドを使用しました

// Set your connectionstring and execute the query and fill your data here 

これは基本的なものです： は、私がコメントを残してきたビットを入力してください。スレッドを追跡し、スレッド信号などを使用して、より高度な振る舞いを実装することで、これを拡張することができます。

さらに、作業を行うコードに余分なパラメータを渡す場合は、これらを作業項目定義クラスに追加します。

注：これは、主なRunParallelメソッドの複数の並列実行をサポートしていませんが、簡単に拡張することができます。

public static class RunParallel 
{ 
    const int NumThreadsToRunInParallel = 8;// Tune this for your DB server performance characteristics 
    public static void FillDataSet(ref DataSet Source) 
    { 
     WorkItemDefinition Work; 
     foreach (DataRow r in Source.Tables["queries"].Rows) 
     { 
      Work = new WorkItemDefinition(); 
      Work.Query = r["QueryStatement"].ToString(); 
      Work.QSource = r["QuerySource"].ToString(); 
      Work.TableName = r["TableName"].ToString(); 
      EnQueueWork(Work); 
     } 
     System.Threading.ThreadStart NewThreadStart; 
     NewThreadStart = new System.Threading.ThreadStart(ProcessPendingWork); 
     for (int I = 0; I < NumThreadsToRunInParallel; I ++) 
     { 
      System.Threading.Thread NewThread; 
      NewThread = new System.Threading.Thread(NewThreadStart); 
      //NewThread.IsBackground = true; //Do this if you want to allow the application to quit before these threads finish all their work and exit 
      ThreadCounterInc(); 
      NewThread.Start(); 
     } 
     while (ThreadCounterValue > 0) 
     { 
      System.Threading.Thread.Sleep(1000); 
     } 
    } 

    private static void ProcessPendingWork() 
    { 
     try 
     { 
      WorkItemDefinition Work; 
      Work = DeQueueWork(); 
      while (Work != null) 
      { 
       Work = DeQueueWork(); 
       DbConnection db = new OdbcConnection(); 
       // Set your connectionstring and execute the query and fill your data here 
      } 
     } 
     finally 
     { 
      ThreadCounterDec(); 
     } 
    } 

    private static int ThreadCounter = 0; 
    private static void ThreadCounterInc() 
    { 
     lock(SyncRoot) 
     { 
      ThreadCounter += 1; 
     } 
    } 
    private static void ThreadCounterDec() 
    { 
     lock (SyncRoot) 
     { 
      ThreadCounter -= 1; 
     } 
    } 
    private static int ThreadCounterValue 
    { 
     get 
     { 
      lock (SyncRoot) 
      { 
       return ThreadCounter; 
      } 
     } 
    } 

    private static object SyncRoot = new object(); 
    private static Queue<WorkItemDefinition> m_PendingWork = new Queue<WorkItemDefinition>(); 
    private static Queue<WorkItemDefinition> PendingWork 
    { 
     get 
     { 
      return m_PendingWork; 
     } 
    } 

    private static WorkItemDefinition DeQueueWork() 
    { 
     lock (SyncRoot) 
     { 
      if (PendingWork.Count > 0) // Catch exception overhead is higher 
      { 
       return PendingWork.Dequeue(); 
      } 
     } 
     return null; 
    } 

    private static void EnQueueWork(WorkItemDefinition Work) 
    { 
     lock (SyncRoot) 
     { 
      PendingWork.Enqueue(Work); 
     } 
    } 

    public class WorkItemDefinition 
    { 
     public string Query { get; set; } 
     public string QSource { get; set; } 
     public string TableName { get; set; } 
    } 
}