20

キューを操作する最良の方法は何かを理解しようとしています。DataTable を返すプロセスがあります。次に、各 DataTable が前の DataTable とマージされます。1 つの問題があります。最後の BulkCopy (OutOfMemory) まで保持するにはレコードが多すぎます。

したがって、受信する各 DataTable をすぐに処理する必要があると判断しました。ConcurrentQueue<T>...について考えてみますが、WriteQueuedData()メソッドがテーブルをデキューしてデータベースに書き込む方法を知る方法がわかりません。

例えば：

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);
    }
}

私の最初の質問は、サブスクライブするイベントが実際にないという事実を除けば、ExtractData()非同期で呼び出す場合、これだけで十分でしょうか? 第二に、機能の仕方について何か欠けていConcurrentQueue<T>て、キューに入れられたオブジェクトと非同期に動作するために何らかの形のトリガーが必要ですか?

更新ConcurrentQueue<T>OnItemQueued イベント ハンドラーを持つ クラスを派生させました。それで：

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);
    }
}

この実装に関する懸念事項はありますか?

4

3 に答える 3

26

問題の私の理解から、あなたはいくつかのことを見逃しています。

並行キューは、データ構造を明示的にロックする必要なく、複数のスレッドがキューに対して読み書きできるように設計されたデータ構造です。(すべてのジャズは舞台裏で処理されます。または、ロックを取得する必要がないようにコレクションが実装されます。)

それを念頭に置いて、使用しようとしているパターンは「プロデュース/コンシューマー」のようです。まず、作業を生成する (およびキューに項目を追加する) いくつかのタスクがあります。次に、キューからのものを消費する (およびアイテムをデキューする) という 2 番目のタスクがあります。

したがって、実際には 2 つのスレッドが必要です。1 つはアイテムを追加し、もう 1 つはアイテムを削除します。並行コレクションを使用しているため、複数のスレッドで項目を追加したり、複数のスレッドで項目を削除したりできます。しかし、明らかに、並行キューで競合が多ければ多いほど、ボトルネックになる速度が速くなります。

于 2010-12-29T02:49:30.607 に答える
23

ConcurrentQueueごく少数の場合にのみ役立つと思います。その主な利点は、ロックフリーであることです。ただし、通常、生産者スレッドは、処理できるデータがあることを何らかの方法で消費者スレッドに通知する必要があります。スレッド間のこのシグナル伝達にはロックが必要であり、 を使用する利点が無効になりConcurrentQueueます。スレッドを同期する最速の方法はMonitor.Pulse()、ロック内でのみ機能する を使用することです。他のすべての同期ツールはさらに低速です。

もちろん、コンシューマはキューに何かがあるかどうかを継続的にチェックできます。これはロックなしで機能しますが、プロセッサ リソースを大量に浪費します。消費者がチェックの間に待機する場合は、少し良くなります。

キューへの書き込み時にスレッドを上げることは、非常に悪い考えです。を使用ConcurrentQueueして保存すると、おそらく 1 マイクロ秒が完全に無駄になりeventhandler、1000 倍の時間がかかる可能性があります。

すべての処理がイベント ハンドラーまたは非同期呼び出しで行われる場合、問題は、なぜキューがまだ必要なのかということです。データをハンドラーに直接渡し、キューをまったく使用しない方がよいでしょう。

ConcurrentQueueの実装は、同時実行を許可するためにかなり複雑であることに注意してください。ほとんどの場合、通常の方法を使用Queue<>して、キューへのすべてのアクセスをロックすることをお勧めします。キューへのアクセスにはマイクロ秒しか必要ないため、2 つのスレッドが同じマイクロ秒でキューにアクセスすることはほとんどなく、ロックによる遅延はほとんどありません。ロックで法線Queue<>を使用すると、多くの場合、 よりもコード実行が速くなりますConcurrentQueue。

于 2014-07-18T08:22:26.480 に答える
3

これは私が思いついたものの完全な解決策です:

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
}

概念実証として、それはかなりうまくいくようです。多くて 4 つのワーカー スレッドが表示されました。

于 2010-12-29T06:03:15.500 に答える