Ben kuyrukta ile çalışmanın en iyi yolu ne olacağını anlamaya çalışıyorum

Question

ConcurrentQueue ile diş nasıl çalışılır. DataTable döndüren bir işlem var. Her bir DataTable, sırayla, önceki DataTable ile birleştirilir. Bir sorun var, son BulkCopy'ye (OutOfMemory) kadar tutmak için çok fazla kayıt var.

Yani, ben hemen her gelen DataTable işlemek gerektiğini belirledi. ConcurrentQueue<T> düşünmek ... ama WriteQueuedData() yöntem bir tablo sıradan çıkarma ve veritabanına yazmak için nasıl tanımış görmüyorum. Örneğin

:

public class TableTransporter 
{ 
    private ConcurrentQueue<DataTable> tableQueue = new ConcurrentQueue<DataTable>(); 

    public TableTransporter() 
    { 
     tableQueue.OnItemQueued += new EventHandler(WriteQueuedData); // no events available 
    } 

    public void ExtractData() 
    { 
     DataTable table; 

     // perform data extraction 
     tableQueue.Enqueue(table); 
    } 

    private void WriteQueuedData(object sender, EventArgs e) 
    { 
     BulkCopy(e.Table); 
    } 
} 

ilk sorum, bir kenara ben ExtractData() ararsanız Aslında abone için bir olay yok olmasından uyumsuz buna ihtiyacım hepsi bu nedir? İkincisi, ConcurrentQueue<T> işlevlerinin yolunda eksik olan ve sıraya alınmış nesnelerle eşzamansız olarak çalışmak için bir tür tetikleyici gerektiren bir şey var mı?

Güncelleme Ben sadece bir OnItemQueued olay işleyicisi vardır ConcurrentQueue<T> gelen bir sınıf elde var. Sonra:

new public void Enqueue (DataTable Table) 
{ 
    base.Enqueue(Table); 
    OnTableQueued(new TableQueuedEventArgs(Table)); 
} 

public void OnTableQueued(TableQueuedEventArgs table) 
{ 
    EventHandler<TableQueuedEventArgs> handler = TableQueued; 

    if (handler != null) 
    { 
     handler(this, table); 
    } 
} 

Bu uygulama ile ilgili herhangi bir endişeniz var mı? Sorunun Benim anlayış

Answer 1

, bir kaç şey eksik.

eşzamanlı kuyruk birden konuları okuma ve açıkça veri yapısını kilitlemek gerek kalmadan kuyruğa yazma kabul etmek üzere tasarlanmış bir veri yapısıdır. (Tüm bu caz sahnelerin ardında halledilir, ya da koleksiyon bir kilit gerektirmeyecek şekilde uygulanır.)

Bunu göz önünde bulundurarak, denediğiniz modele benziyor. Kullanmak "Üretin/Tüketici" dir. Öncelikle, iş üreten bazı görevleriniz var (ve sıraya öğeler ekleyerek). İkincisi ise ikinci bir göreviniz var.

Gerçekten iki parçayı da istiyorsunuz: bir tane öğe ve ikinci bir çıkarma öğesi. Eşzamanlı bir koleksiyon kullandığınız için, öğelerin eklenmesi ve öğelerin kaldırılmasında birden fazla iş parçacığı ekleyerek birden çok iş parçacığınız olabilir. Ama tabii ki eşzamanlı sıradaki daha fazla çekişme, darboğaz haline gelen daha hızlıdır.

Answer 2

Bu benim ile geldi ne için komple bir çözümdür:

public class TableTransporter 
{ 
    private static int _indexer; 

    private CustomQueue tableQueue = new CustomQueue(); 
    private Func<DataTable, String> RunPostProcess; 
    private string filename; 

    public TableTransporter() 
    { 
     RunPostProcess = new Func<DataTable, String>(SerializeTable); 
     tableQueue.TableQueued += new EventHandler<TableQueuedEventArgs>(tableQueue_TableQueued); 
    } 

    void tableQueue_TableQueued(object sender, TableQueuedEventArgs e) 
    { 
     // do something with table 
     // I can't figure out is how to pass custom object in 3rd parameter 
     RunPostProcess.BeginInvoke(e.Table,new AsyncCallback(PostComplete), filename); 
    } 

    public void ExtractData() 
    { 
     // perform data extraction 
     tableQueue.Enqueue(MakeTable()); 
     Console.WriteLine("Table count [{0}]", tableQueue.Count); 
    } 

    private DataTable MakeTable() 
    { return new DataTable(String.Format("Table{0}", _indexer++)); } 

    private string SerializeTable(DataTable Table) 
    { 
     string file = Table.TableName + ".xml"; 

     DataSet dataSet = new DataSet(Table.TableName); 

     dataSet.Tables.Add(Table); 

     Console.WriteLine("[{0}]Writing {1}", Thread.CurrentThread.ManagedThreadId, file); 
     string xmlstream = String.Empty; 

     using (MemoryStream memstream = new MemoryStream()) 
     { 
      XmlSerializer xmlSerializer = new XmlSerializer(typeof(DataSet)); 
      XmlTextWriter xmlWriter = new XmlTextWriter(memstream, Encoding.UTF8); 

      xmlSerializer.Serialize(xmlWriter, dataSet); 
      xmlstream = UTF8ByteArrayToString(((MemoryStream)xmlWriter.BaseStream).ToArray()); 

      using (var fileStream = new FileStream(file, FileMode.Create)) 
       fileStream.Write(StringToUTF8ByteArray(xmlstream), 0, xmlstream.Length + 2); 
     } 
     filename = file; 

     return file; 
    } 

    private void PostComplete(IAsyncResult iasResult) 
    { 
     string file = (string)iasResult.AsyncState; 
     Console.WriteLine("[{0}]Completed: {1}", Thread.CurrentThread.ManagedThreadId, file); 

     RunPostProcess.EndInvoke(iasResult); 
    } 

    public static String UTF8ByteArrayToString(Byte[] ArrBytes) 
    { return new UTF8Encoding().GetString(ArrBytes); } 

    public static Byte[] StringToUTF8ByteArray(String XmlString) 
    { return new UTF8Encoding().GetBytes(XmlString); } 
} 

public sealed class CustomQueue : ConcurrentQueue<DataTable> 
{ 
    public event EventHandler<TableQueuedEventArgs> TableQueued; 

    public CustomQueue() 
    { } 
    public CustomQueue(IEnumerable<DataTable> TableCollection) 
     : base(TableCollection) 
    { } 

    new public void Enqueue (DataTable Table) 
    { 
     base.Enqueue(Table); 
     OnTableQueued(new TableQueuedEventArgs(Table)); 
    } 

    public void OnTableQueued(TableQueuedEventArgs table) 
    { 
     EventHandler<TableQueuedEventArgs> handler = TableQueued; 

     if (handler != null) 
     { 
      handler(this, table); 
     } 
    } 
} 

public class TableQueuedEventArgs : EventArgs 
{ 
    #region Fields 
    #endregion 

    #region Init 
    public TableQueuedEventArgs(DataTable Table) 
    {this.Table = Table;} 
    #endregion 

    #region Functions 
    #endregion 

    #region Properties 
    public DataTable Table 
    {get;set;} 
    #endregion 
} 

kavram ispatı olarak, oldukça iyi iş gibi görünüyor. En çok 4 işçi iş parçacığı gördüm.

Answer 3

ConcurrentQueue'nun yalnızca birkaç durumda yararlı olduğunu düşünüyorum. Ana avantajı, kilitlenmemesidir. Bununla birlikte, genellikle üretici iplik (ler), tüketici ipliğini bir şekilde işlemek için elverişli veri olduğunu bildirmek zorundadır. İş parçacığı arasındaki bu sinyalizasyon kilitlere ihtiyaç duyar ve ConcurrentQueue kullanmanın yararını ortadan kaldırır. İş parçacığı senkronize etmenin en hızlı yolu, yalnızca bir kilit içinde çalışan Monitor.Pulse() yöntemini kullanmaktır. Diğer tüm senkronizasyon araçları daha yavaştır. Tabii ki, tüketiciler, kuyrukta çalışacak, ancak işlemci kaynaklarının büyük bir israfı olan kuyrukta bir şey olup olmadığını sürekli kontrol edebilirler. Tüketici kontrol arasında beklerse biraz daha iyi olur.

Sıraya yazarken bir iş parçacığı oluşturmak çok kötü bir fikirdir. Mabe 1 mikro saniyesini kaydetmek için ConcurrentQueue kullanılarak, 1000 kat daha uzun sürebilir, eventhandler çalıştırılarak tamamen boşa harcanacaktır.

Tüm işlem bir olay işleyicisinde veya bir zaman uyumsuz çağrısında gerçekleştirilirse, sorun hala bir sıraya ihtiyaç duyuyor mu? Verileri doğrudan işleyiciye iletin ve bir sıra kullanmayın.

ConcurrentQueue uygulamasının eşzamanlılık sağlamak için oldukça karmaşık olduğunu lütfen unutmayın. Çoğu durumda, normal bir Kuyruk <> öğesini daha iyi kullanın ve kuyruğa her erişimi kilitleyin. Kuyruk girişinin yalnızca mikrosaniye gerektirdiğinden, 2 iş parçacığının aynı mikrosaniye içinde kuyruğa erişmesi çok olası değildir ve kilitleme nedeniyle neredeyse hiç gecikme olmayacaktır. Kilitleme ile normal bir Queue <> kullanma ConcurrentQueue'den daha hızlı kod yürütme ile sonuçlanır.