如何在TPL中實現與BlockingCollection類似的功能？

我創建了一個包含BlockingCollection的簡單類。它代表將按照接收順序執行的一系列操作。我已經閱讀了很多關於TPL的文章，似乎我應該使用它而不是我目前使用的文章。原因之一是單元測試會更容易，代碼也會寫得更少。我知道你可以使用Task.Factory.StartNew（）等方法輕鬆完成新的任務，但不知道如何以類似的方式使用它，就像我現在擁有的類一樣。我如何用TPL完成同樣的事情？如何在TPL中實現與BlockingCollection類似的功能？

按照要求，這裏是我創建的類：創建簡單的生產者 - 消費者的場景

public class MyService 
{ 
    /// <summary>Queue of actions to be consumed on a separate thread</summary> 
    private BlockingCollection<MyObject> queue = new BlockingCollection<MyObject>(); 

    public MyService() 
    { 
     StartService(); 
    } 

    public void AddToQueue(MyObject newObject) 
    { 
     queue.Add(newObject); 
    } 

    private void StartService() 
    { 
     System.Threading.Tasks.Task.Factory.StartNew(() => 
     { 
      while (true) 
      { 
       try 
       { 
        MyObject myObject = queue.Take(); // blocks until new object received 

        // Do work... 
       } 
       catch (Exception e) 
       { 
        // Log... 
       } 
      } 
     }); 
    } 
}

來源

2016-02-22 Andrew

如果你沒有正確地使用'StartNew'，請小心[當你不希望它在UI線程上運行時]（http://blog.stephencleary.com/2013/ 08/startnew-is-dangerous.html），使用'Task.Run（'，除非你有特定的理由使用'StartNew'。另外，給你一個總體的方向推動（我不知道這是否是你需要的），但是請看[TPL DataFlow]（https://msdn.microsoft.com/en-us/library/hh228603（v = vs.110）.aspx）。 –

你可以給出一個代碼示例你想要做什麼？ – Coding4Fun

我剛剛添加了代碼，以使其更清晰。 – Andrew

BlockingCollection和異步藏品家庭。（例如一個作家和多個閱讀器）

當然 - 你可以設法建立幾乎相同的Task.Run，將添加，刪除，清理等項目非同步收集像List<T>，但你必須管理所有多線程問題（而且還有很多問題）。

例如：

public class MyService 
{ 
    /// <summary>Queue of actions to be consumed on a separate thread</summary> 
    private BlockingCollection<MyObject> queue = new BlockingCollection<MyObject>(); 
    private IEnumerable<Task> readers = Enumerable.Range(0, 10).Select((t) => new Task(() => this.StartService())); 

public MyService() 
{ 
    StartService(); 
    readers.AsParallel().ForAll(t => t.Start()); 
} 

public void AddToQueue(MyObject newObject) 
{ 
    queue.Add(newObject); 
} 

private void StartService() 
{ 
     while (true) 
     { 
      try 
      { 
       MyObject myObject = queue.Take(); // blocks until new object received 

       // Do work... 
      } 
      catch (Exception e) 
      { 
       // Log... 
      } 
     } 
    } 
}

你看 - 有一個在同一個集合多個 '讀者'。如果你自己完成了BlockingCollection你應該處理所有lock s收集等。

來源

2016-02-22 15:43:12 Jasper

那麼最好在這種情況下使用BlockingCollection呢？ – Andrew

@Andrew是的。這個答案爲您提供了多個消費者的示例，您的代碼只有一個消費者。如果你只需要一個消費者，你的代碼就好了。如果您想關閉服務，可能是取消令牌。 – Maarten

它建議在多線程場景中使用BlockingCollection（和變體） – Jasper

舊時尚，阻止同步和基於任務的異步混合不好。

Task.Run(() => 
    { 
     while (true) 
     { 
      // some thing that sometimes blocks 
     } 
    });

只寫

new Thread(() => 
    { 
     while (true) 
     { 
      // some thing that sometimes blocks 
     } 
});

這些都將佔用一個線程幾乎永遠的幻想方式。第一個將使用線程池中的一個，它應該比專門創建的更好，但由於它以後從未發佈，優點消失。

如果你想使用任務和TPL，並從中受益，你應該儘可能避免阻塞任何東西。你可以例如使用ConcurrentQueue作爲後盾隊列，做這樣的事情：

public class MyService 
{ 
    /// <summary>Queue of actions to be consumed by separate task</summary> 
    private ConcurrentQueue<MyObject> queue = new ConcurrentQueue<MyObject>(); 

    private bool _isRunning = false; 
    private Task _consumingTask; 

    public MyService() 
    { 
    } 

    public void AddToQueue(MyObject newObject) 
    { 
     queue.Add(newObject); 
    } 

    private void StartService() 
    { 
     _isRunning = true; 
     Task.Run(async() => 
     { 
      while (_isRunning) 
      { 
       MyObject myObject; 

       while(_isRunning && queue.TryDequeue(out myObject) 
       { 
        try 
        { 
         // Do work... 
        } 
        catch (Exception e) 
        { 
         // Log... 
        } 
       } 
       await Task.Delay(100); // tune this value to one pertinent to your use case 
      } 
     }); 
    } 

    public void StopService() 
    { 
     _isRunning = false; 
     _consumingTask.Wait(); 
    } 
}

此實現永不塊，只佔用一個線程在真正需要的時候計算。它也很容易與其他Task優雅地混合在一起。

TLDR：如果你去Task的方式，一路走。中間點實際上不是你想要的地方，你會得到所有的複雜性和優點。

來源

2016-02-22 16:14:43 Falanwe

如何在TPL中實現與BlockingCollection類似的功能？

回答

相關問題