5

我正在对自己进行有关 Parallel.Invoke 和一般并行处理的教育，以便在当前项目中使用。我需要朝着正确的方向努力，以了解如何根据需要动态\智能地分配更多并行“线程”。

举个例子。假设您正在解析大型日志文件。这涉及从文件中读取，对返回的行进行某种解析，最后写入数据库。

所以对我来说，这是一个可以从并行处理中受益的典型问题。

作为简单的第一遍，以下代码实现了这一点。

Parallel.Invoke(
  ()=> readFileLinesToBuffer(),
  ()=> parseFileLinesFromBuffer(),
  ()=> updateResultsToDatabase()    
);

在幕后

  1. readFileLinesToBuffer() 读取每一行并存储到缓冲区。
  2. parseFileLinesFromBuffer 出现并使用缓冲区中的行，然后假设它将它们放在另一个缓冲区上，以便 updateResultsToDatabase() 可以出现并使用此缓冲区。

因此，显示的代码假定三个步骤中的每一个都使用相同的时间\资源，但是可以说 parseFileLinesFromBuffer() 是一个长时间运行的过程，因此您希望并行运行两个方法，而不是只运行其中一个方法。

您如何让代码根据它可能感知到的任何瓶颈智能地决定执行此操作？

从概念上讲，我可以看到一些监视缓冲区大小的方法可能如何工作，例如产生一个新的“线程”以增加速率消耗缓冲区......但我认为在组合 TPL 库时已经考虑了这种类型的问题.

一些示例代码会很棒，但我真的只需要一个线索来了解我接下来应该研究哪些概念。看起来可能 System.Threading.Tasks.TaskScheduler 掌握着关键？

4

3 回答 3

4

您是否尝试过响应式扩展？

http://msdn.microsoft.com/en-us/data/gg577609.aspx

Rx 是 Microsoft 的一项新技术，其重点如官方网站所述：

Reactive Extensions (Rx)... ...是一个使用可观察集合和 LINQ 样式查询运算符组成异步和基于事件的程序的库。

您可以将其作为 Nuget 包下载

https://nuget.org/packages/Rx-Main/1.0.11226

由于我目前正在学习 Rx，所以我想举这个例子，只是为它编写代码，我最终得到的代码实际上并不是并行执行的，而是完全异步的，并且保证了源代码行是按顺序执行的。

也许这不是最好的实现，但就像我说的我正在学习 Rx，（线程安全应该是一个很好的改进）

这是我用来从后台线程返回数据的 DTO

class MyItem
{
    public string Line { get; set; }
    public int CurrentThread { get; set; }
}

这些是做实际工作的基本方法，我用一个简单的方法来模拟时间，Thread.Sleep我正在返回用于执行每个方法的线程Thread.CurrentThread.ManagedThreadId。注意它的定时器ProcessLine是4秒，这是最耗时的操作

private IEnumerable<MyItem> ReadLinesFromFile(string fileName)
{
    var source = from e in Enumerable.Range(1, 10)
                 let v = e.ToString()
                 select v;

    foreach (var item in source)
    {
        Thread.Sleep(1000);
        yield return new MyItem { CurrentThread = Thread.CurrentThread.ManagedThreadId, Line = item };
    }
}

private MyItem UpdateResultToDatabase(string processedLine)
{
    Thread.Sleep(700);
    return new MyItem { Line = "s" + processedLine, CurrentThread = Thread.CurrentThread.ManagedThreadId };
}

private MyItem ProcessLine(string line)
{
    Thread.Sleep(4000);
    return new MyItem { Line = "p" + line, CurrentThread = Thread.CurrentThread.ManagedThreadId };
}

以下方法我使用它只是为了更新 UI

private void DisplayResults(MyItem myItem, Color color, string message)
{
    this.listView1.Items.Add(
        new ListViewItem(
            new[]
            {
                message, 
                myItem.Line ,
                myItem.CurrentThread.ToString(), 
                Thread.CurrentThread.ManagedThreadId.ToString()
            }
        )
        {
            ForeColor = color
        }
    );
}

最后这是调用 Rx API 的方法

private void PlayWithRx()
{
    // we init the observavble with the lines read from the file
    var source = this.ReadLinesFromFile("some file").ToObservable(Scheduler.TaskPool);

    source.ObserveOn(this).Subscribe(x =>
    {
        // for each line read, we update the UI
        this.DisplayResults(x, Color.Red, "Read");

        // for each line read, we subscribe the line to the ProcessLine method
        var process = Observable.Start(() => this.ProcessLine(x.Line), Scheduler.TaskPool)
            .ObserveOn(this).Subscribe(c =>
            {
                // for each line processed, we update the UI
                this.DisplayResults(c, Color.Blue, "Processed");

                // for each line processed we subscribe to the final process the UpdateResultToDatabase method
                // finally, we update the UI when the line processed has been saved to the database
                var persist = Observable.Start(() => this.UpdateResultToDatabase(c.Line), Scheduler.TaskPool)
                    .ObserveOn(this).Subscribe(z => this.DisplayResults(z, Color.Black, "Saved"));
            });
    });
}

此过程完全在后台运行，这是生成的输出：

在此处输入图像描述

于 2012-05-31T04:45:57.367 回答
0

在 async/await 世​​界中，你会有类似的东西：

public async Task ProcessFileAsync(string filename)
{
    var lines = await ReadLinesFromFileAsync(filename);
    var parsed = await ParseLinesAsync(lines);
    await UpdateDatabaseAsync(parsed);
}

然后调用者可以做 var tasks = filenames.Select(ProcessFileAsync).ToArray(); 和任何东西（WaitAll、WhenAll 等，取决于上下文）

于 2012-05-31T05:19:24.853 回答
0

使用几个BlockingCollection. 这是一个例子

这个想法是您创建一个producer将数据放入集合中的

while (true) {
    var data = ReadData();
    blockingCollection1.Add(data);
}

然后创建任意数量的从集合中读取的消费者

while (true) {
    var data = blockingCollection1.Take();
    var processedData = ProcessData(data);
    blockingCollection2.Add(processedData);
}

等等

您还可以使用 Parallel.Foreach 让 TPL 处理消费者的数量

Parallel.ForEach(blockingCollection1.GetConsumingPartitioner(),
                 data => {
                          var processedData = ProcessData(data);
                          blockingCollection2.Add(processedData);
                 });

（请注意，您需要使用GetConsumingPartitionernot GetConsumingEnumerable（请参见此处）

于 2012-05-31T09:33:25.580 回答