Модель Consumer производителя с использованием TPL, Задачи в .net 4.0 - PullRequest
Новая
       24

Модель Consumer производителя с использованием TPL, Задачи в .net 4.0

0 голосов
/ 17 октября 2011

У меня довольно большой XML-файл (около 1-2 ГБ).

Требуется сохранить данные XML в базе данных. В настоящее время это достигается в 3 этапа.

  1. Чтение большого файла с минимальным отпечатком памяти, насколько это возможно
  2. Создание сущностей из xml-данных
  3. Сохранение данных из созданных объектов в базе данных с помощью SqlBulkCopy.

Для достижения лучшей производительности я хочу создать модель «производитель-потребитель», в которой производитель создает набор сущностей, скажем, по 10 КБ, и добавляет его в очередь. А потребитель должен взять пакет сущностей из очереди и сохранить его в базе данных с помощью sqlbulkcopy.

Спасибо, Gokul 

void Main()
{
    int iCount = 0;
    string fileName = @"C:\Data\CatalogIndex.xml";

    DateTime startTime = DateTime.Now;
    Console.WriteLine("Start Time: {0}", startTime);
    FileInfo fi = new FileInfo(fileName);
    Console.WriteLine("File Size:{0} MB", fi.Length / 1048576.0);

/* I want to change this loop to create a producer consumer pattern here to process the data parallel-ly
*/
     foreach (var element in StreamElements(fileName,"title"))
            {
                iCount++;
            }

            Console.WriteLine("Count: {0}", iCount);
            Console.WriteLine("End Time: {0}, Time Taken:{1}", DateTime.Now, DateTime.Now - startTime);
        }

    private static IEnumerable<XElement> StreamElements(string fileName, string elementName)
    { 
        using (var rdr = XmlReader.Create(fileName))
        {
            rdr.MoveToContent();
            while (!rdr.EOF)
            {
                if ((rdr.NodeType == XmlNodeType.Element) && (rdr.Name == elementName))
                {
                    var e = XElement.ReadFrom(rdr) as XElement;
                    yield return e;
                }
                else
                {
                    rdr.Read();
                }
            }
            rdr.Close();
        }
    }

1 Ответ

4 голосов
/ 17 октября 2011

Это то, что вы пытаетесь сделать? 

    void Main()
    {
        const int inputCollectionBufferSize = 1024;
        const int bulkInsertBufferCapacity = 100;
        const int bulkInsertConcurrency = 4;

        BlockingCollection<object> inputCollection = new BlockingCollection<object>(inputCollectionBufferSize);

        Task loadTask = Task.Factory.StartNew(() =>
        {
            foreach (object nextItem in ReadAllElements(...))
            {
                // this will potentially block if there are already enough items
                inputCollection.Add(nextItem);
            }

            // mark this collection as done
            inputCollection.CompleteAdding();
        });

        Action parseAction = () =>
        {
            List<object> bulkInsertBuffer = new List<object>(bulkInsertBufferCapacity);

            foreach (object nextItem in inputCollection.GetConsumingEnumerable())
            {
                if (bulkInsertBuffer.Length == bulkInsertBufferCapacity)
                {
                    CommitBuffer(bulkInsertBuffer);
                    bulkInsertBuffer.Clear();
                }

                bulkInsertBuffer.Add(nextItem);
            }
        };

        List<Task> parseTasks = new List<Task>(bulkInsertConcurrency);

        for (int i = 0; i < bulkInsertConcurrency; i++)
        {
            parseTasks.Add(Task.Factory.StartNew(parseAction));
        }

        // wait before exiting
        loadTask.Wait();
        Task.WaitAll(parseTasks.ToArray());
    }
...