ExecureSegmentedQueryとStreamWriterのAzureテーブルストレージ分離スレッド

Question

私はAzureテーブルストレージからデータを取得し、それをテキストファイルに転送するコンソールアプリケーションを持っています。今、このコードは、テキストファイルに書き込む前に最初の1000件のレコードを取得することを意味し、シングルスレッドで実行

using (StreamWriter writer = new StreamWriter(filePath)) 
     { 
      //header 
      writer.Write("PartitionKey|RowKey|Timestamp|Id|"); 
      writer.WriteLine(); 

      do 
      { 
       var entities = new List<TModel>(); 
       var queryResult = table.ExecuteQuerySegmented(new TableQuery<TModel>(), token); 
       entities.AddRange(queryResult.Results); 
       count = count + 1000; 

       Console.WriteLine("{0} records retrieved", count); 

       //records 
       foreach (TModel entity in entities) 
       { 
        writer.Write("\"" + entity.PartitionKey + "\"|\"" + entity.RowKey + "\"|" + entity.Timestamp + "|" + entity.Id); 
        writer.WriteLine(); 
       } 
       token = queryResult.ContinuationToken; 
      } 
      while (token != null); 
      writer.Dispose(); 

は、次の1000のレコードは、その後追加取得します：現在、私のコードは次のようになりそれをファイルに保存します。

データ検索の処理とファイルへの書き込みを2つの別々のスレッドに分割する方法があるとすれば、スレッドがセグメント化されたクエリの実行を実行するだけで、ファイルへのデータ（キューのような）？

ご提案は大歓迎です。前もって感謝します！

Answer 1

説明に基づいて、2つのスレッドを作成することをお勧めします。 1つはプロデューサスレッドで、もう1つはコンシューマスレッドです。さらに、スレッドセーフなコレクションが必要です。 ConcurrentQueueクラスを使用して、これらの2つのスレッドからアクセスされるエンティティを格納できます。

下記のコードは参考用です。

class Program 
{ 
    static void Main(string[] args) 
    { 
     //Run producer thread 
     Task.Run(new Action(ReadDataFromTable)); 
     //Run consumer thread 
     Task.Run(new Action(WriteDataToText)); 

     Console.Read(); 
    } 

    public static ConcurrentQueue<TModel> entitiesQueue = new ConcurrentQueue<TModel>(); 
    //This tag point out whether the read process is down 
    public static bool ReadTableFinished = false; 
    public static void ReadDataFromTable() 
    { 
     do 
     { 
      var entities = new List<TModel>(); 
      var queryResult = table.ExecuteQuerySegmented(new TableQuery<TModel>(), token); 
      entities.AddRange(queryResult.Results); 
      count = count + 1000; 

      Console.WriteLine("{0} records retrieved", count); 

      //Save entites to the queue 
      foreach (TModel entity in entities) 
      { 
       entitiesQueue.Enqueue(entity); 
      } 
      token = queryResult.ContinuationToken; 
     } 
     while (token != null); 

     ReadTableFinished = true; 
    } 

    public static void WriteDataToText() 
    { 
     using (StreamWriter writer = new StreamWriter(filePath)) 
     { 
      //header 
      writer.Write("PartitionKey|RowKey|Timestamp|Id|"); 
      writer.WriteLine(); 

      do 
      { 
       TModel entity = null; 
       //Read and remove entity from queue 
       entitiesQueue.TryDequeue(out entity); 

       if (entity != null) 
       { 
        //Write data to text file 
        writer.Write("\"" + entity.PartitionKey + "\"|\"" + entity.RowKey + "\"|" + entity.Timestamp + "|" + entity.Id); 
        writer.WriteLine(); 
       } 
       else 
       { 
        //If all the entities are read out from the table and now there is no entities in the queue, we will break this loop and exit current thread. 
        if (ReadTableFinished) 
        { 
         break; 
        } 
       } 
      } 
      while (true); 
      writer.Dispose(); 
     } 
    } 
} 

ブレークポイントを追加する）（Console.Readずにコードを入力しないであろう。

UIスレッドをブロックするConsole.Read（）を追加しないと、UIスレッドは完全に実行されます。 UIスレッドの状態が完了していれば、他のスレッド（タスク）は実行できません。 Console.Read（）を追加したくない場合。 Mainメソッドで次のコードを代わりに使用できます。 Task.WaitAllメソッドは、すべてのタスクが完全に実行されるまで、現在のUIスレッドをブロックします。

static void Main(string[] args) 
{ 
    //Run producer thread 
    Task t1 = Task.Run(new Action(ReadDataFromTable)); 
    //Run consumer thread 
    Task t2 = Task.Run(new Action(WriteDataToText)); 

    Task.WaitAll(t1, t2); 
    Console.WriteLine("all completed"); 
}