BlockingCollection 与异步任务
BlockingCollection with async task
我正在尝试正确建模一个多线程 single-producer/multi-consumer 场景，在该场景中，消费者可以要求生产者获取物品，但生产者需要执行耗时的操作来生产它（想想执行查询或打印文档）。
我的目标是确保没有消费者可以同时要求生产者生产一件商品。在我的真实世界用例中，生产者是一个硬件控制器，它必须确保一次只向硬件发送一个请求。其他并发请求最终必须等待或被拒绝（我知道如何拒绝它们，所以让我们集中精力让它们等待）。
我希望生产者和每个消费者在不同的线程中运行。
我无法仅使用 BlockingCollection 获得干净的代码。我必须将它与 SemaphoreSlim 一起使用，否则消费者可能会出现竞争条件。
我 认为 它应该可以工作（事实上它在我所有的测试中都运行良好），即使我不是 100% 确定它。
这是我的程序：
制作人：
class Producer : IDisposable
{
//Explicit waiting item => I feel this should not be there
private SemaphoreSlim _semaphore;
private BlockingCollection<Task<string>> _collection;
public Producer()
{
_collection = new BlockingCollection<Task<string>>(new ConcurrentQueue<Task<string>>(), 1);
_semaphore = new SemaphoreSlim(1, 1);
}
public void Start()
{
Task consumer = Task.Factory.StartNew(() =>
{
try
{
while (!_collection.IsCompleted)
{
Task<string> current = _collection.Take();
current.RunSynchronously(); //Is this bad?
//Signal the long running operation has ended => This is what I'm not happy about
_semaphore.Release();
}
}
catch (InvalidOperationException)
{
Console.WriteLine("Adding was compeleted!");
}
});
}
public string GetRandomString(string consumerName)
{
Task<string> task = new Task<string>(() =>
{
//Simulate long running operation
Thread.Sleep(100);
return GetRandomString();
});
_collection.Add(task);
//Wait for long running operation to complete => This is what I'm not happy about
_semaphore.Wait();
Console.WriteLine("Producer produced {0} by {1} request", task.Result, consumerName);
return task.Result;
}
public void Dispose()
{
_collection.CompleteAdding();
}
private string GetRandomString()
{
var chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
var random = new Random();
var result = new string(Enumerable
.Repeat(chars, 8)
.Select(s => s[random.Next(s.Length)])
.ToArray());
return result;
}
}
消费者：
class Consumer
{
Producer _producer;
string _name;
public Consumer(
Producer producer,
string name)
{
_producer = producer;
_name = name;
}
public string GetOrderedString()
{
string produced = _producer.GetRandomString(_name);
return String.Join(String.Empty, produced.OrderBy(c => c));
}
}
控制台应用程序：
class Program
{
static void Main(string[] args)
{
int consumerNumber = 5;
int reps = 10;
Producer prod = new Producer();
prod.Start();
Task[] consumers = new Task[consumerNumber];
for (var cConsumers = 0; cConsumers < consumerNumber; cConsumers++)
{
Consumer consumer = new Consumer(prod, String.Format("Consumer{0}", cConsumers + 1));
Task consumerTask = Task.Factory.StartNew((consumerIndex) =>
{
int cConsumerNumber = (int)consumerIndex;
for (var counter = 0; counter < reps; counter++)
{
string data = consumer.GetOrderedString();
Console.WriteLine("Consumer{0} consumed {1} at iteration {2}", cConsumerNumber, data, counter + 1);
}
}, cConsumers + 1);
consumers[cConsumers] = consumerTask;
}
Task continuation = Task.Factory.ContinueWhenAll(consumers, (c) =>
{
prod.Dispose();
Console.WriteLine("Producer/Consumer ended");
Console.ReadLine();
});
continuation.Wait();
}
}
我关心的是这是否是解决问题的正确方法，或者你们是否可以建议任何其他最佳实践。
我已经用谷歌搜索并尝试了不同的建议想法，但我尝试过的每个例子都假设生产者能够在他们被要求后立即生产物品……在现实世界的应用程序中这种情况非常罕见:)
非常感谢任何帮助。
我认为信号量并不是真的需要。 Task class and the available concurrent collections.
应该已经满足了您的所有需求
我试着创建了一些示例代码。不幸的是，我仍然没有完全理解 async/await，所以也许这也是一个对你的情况有帮助的领域（如果你的真正任务主要是 I/O 约束而不是 cpu绑定）。
但是正如您已经看到的，不需要信号量或类似的东西。所有这些事情都是由提供的 类 完成的（例如调用 BlockingCollection.GetConsumingEnumerable()）。
所以希望这能有所帮助。
Program.cs
private static void Main(string[] args)
{
var producer = new Producer();
var consumer = new Consumer(producer.Workers);
consumer.Start();
producer.Start();
Console.ReadKey();
}
Producer.cs
public class Producer : IDisposable
{
private CancellationTokenSource _Cts;
private Random _Random = new Random();
private int _WorkCounter = 0;
private BlockingCollection<Task<String>> _Workers;
private Task _WorkProducer;
public Producer()
{
_Workers = new BlockingCollection<Task<String>>();
}
public IEnumerable<Task<String>> Workers
{
get { return _Workers.GetConsumingEnumerable(); }
}
public void Dispose()
{
Stop();
}
public void Start()
{
if (_Cts != null)
throw new InvalidOperationException("Producer has already been started.");
_Cts = new CancellationTokenSource();
_WorkProducer = Task.Factory.StartNew(() => Run(_Cts.Token));
}
public void Stop()
{
var cts = _Cts;
if (cts != null)
{
cts.Cancel();
cts.Dispose();
_Cts = null;
}
_WorkProducer.Wait();
}
private String GetRandomString()
{
var chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
var result = new String(Enumerable
.Repeat(chars, 8)
.Select(s => s[_Random.Next(s.Length)])
.ToArray());
return result;
}
private void Run(CancellationToken token)
{
while (!token.IsCancellationRequested)
{
var worker = StartNewWorker();
_Workers.Add(worker);
Task.Delay(100);
}
_Workers.CompleteAdding();
_Workers = new BlockingCollection<Task<String>>();
}
private Task<String> StartNewWorker()
{
return Task.Factory.StartNew<String>(Worker);
}
private String Worker()
{
var workerId = Interlocked.Increment(ref _WorkCounter);
var neededTime = TimeSpan.FromSeconds(_Random.NextDouble() * 5);
Console.WriteLine("Worker " + workerId + " starts in " + neededTime);
Task.Delay(neededTime).Wait();
var result = GetRandomString();
Console.WriteLine("Worker " + workerId + " finished with " + result);
return result;
}
}
Consumer.cs
public class Consumer
{
private Task _Consumer;
private IEnumerable<Task<String>> _Workers;
public Consumer(IEnumerable<Task<String>> workers)
{
if (workers == null)
throw new ArgumentNullException("workers");
_Workers = workers;
}
public void Start()
{
var consumer = _Consumer;
if (consumer == null
|| consumer.IsCompleted)
{
_Consumer = Task.Factory.StartNew(Run);
}
}
private void Run()
{
foreach (var worker in _Workers)
{
var result = worker.Result;
Console.WriteLine("Received result " + result);
}
}
}
如果我对你的理解是正确的，你想确保一次只有一个任务被所谓的 "producer" 处理。然后对您的代码稍作修改，您可以这样做：
internal class Producer : IDisposable {
private readonly BlockingCollection<RandomStringRequest> _collection;
public Producer() {
_collection = new BlockingCollection<RandomStringRequest>(new ConcurrentQueue<RandomStringRequest>());
}
public void Start() {
Task consumer = Task.Factory.StartNew(() => {
try {
foreach (var request in _collection.GetConsumingEnumerable()) {
Thread.Sleep(100); // long work
request.SetResult(GetRandomString());
}
}
catch (InvalidOperationException) {
Console.WriteLine("Adding was compeleted!");
}
});
}
public RandomStringRequest GetRandomString(string consumerName) {
var request = new RandomStringRequest();
_collection.Add(request);
return request;
}
public void Dispose() {
_collection.CompleteAdding();
}
private string GetRandomString() {
var chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
var random = new Random();
var result = new string(Enumerable
.Repeat(chars, 8)
.Select(s => s[random.Next(s.Length)])
.ToArray());
return result;
}
}
internal class RandomStringRequest : IDisposable {
private string _result;
private ManualResetEvent _signal;
public RandomStringRequest() {
_signal = new ManualResetEvent(false);
}
public void SetResult(string result) {
_result = result;
_signal.Set();
}
public string GetResult() {
_signal.WaitOne();
return _result;
}
public bool TryGetResult(TimeSpan timeout, out string result) {
result = null;
if (_signal.WaitOne(timeout)) {
result = _result;
return true;
}
return false;
}
public void Dispose() {
_signal.Dispose();
}
}
internal class Consumer {
private Producer _producer;
private string _name;
public Consumer(
Producer producer,
string name) {
_producer = producer;
_name = name;
}
public string GetOrderedString() {
using (var request = _producer.GetRandomString(_name)) {
// wait here for result to be prepared
var produced = request.GetResult();
return String.Join(String.Empty, produced.OrderBy(c => c));
}
}
}
请注意，生产者是单线程的，它使用 GetConsumingEnumerable。也没有信号量，也没有任务。相反，RandomStringRequest 返回给消费者，并且在调用 GetResult 或 TryGetResult 时，它将等待生产者产生结果（或超时到期）。您可能还想在某些地方传递 CancellationTokens（例如 GetConsumingEnumerable）。
我正在尝试正确建模一个多线程 single-producer/multi-consumer 场景，在该场景中，消费者可以要求生产者获取物品，但生产者需要执行耗时的操作来生产它（想想执行查询或打印文档）。
我的目标是确保没有消费者可以同时要求生产者生产一件商品。在我的真实世界用例中，生产者是一个硬件控制器，它必须确保一次只向硬件发送一个请求。其他并发请求最终必须等待或被拒绝（我知道如何拒绝它们，所以让我们集中精力让它们等待）。
我希望生产者和每个消费者在不同的线程中运行。
我无法仅使用 BlockingCollection 获得干净的代码。我必须将它与 SemaphoreSlim 一起使用，否则消费者可能会出现竞争条件。
我 认为 它应该可以工作（事实上它在我所有的测试中都运行良好），即使我不是 100% 确定它。
这是我的程序：
制作人：
class Producer : IDisposable
{
//Explicit waiting item => I feel this should not be there
private SemaphoreSlim _semaphore;
private BlockingCollection<Task<string>> _collection;
public Producer()
{
_collection = new BlockingCollection<Task<string>>(new ConcurrentQueue<Task<string>>(), 1);
_semaphore = new SemaphoreSlim(1, 1);
}
public void Start()
{
Task consumer = Task.Factory.StartNew(() =>
{
try
{
while (!_collection.IsCompleted)
{
Task<string> current = _collection.Take();
current.RunSynchronously(); //Is this bad?
//Signal the long running operation has ended => This is what I'm not happy about
_semaphore.Release();
}
}
catch (InvalidOperationException)
{
Console.WriteLine("Adding was compeleted!");
}
});
}
public string GetRandomString(string consumerName)
{
Task<string> task = new Task<string>(() =>
{
//Simulate long running operation
Thread.Sleep(100);
return GetRandomString();
});
_collection.Add(task);
//Wait for long running operation to complete => This is what I'm not happy about
_semaphore.Wait();
Console.WriteLine("Producer produced {0} by {1} request", task.Result, consumerName);
return task.Result;
}
public void Dispose()
{
_collection.CompleteAdding();
}
private string GetRandomString()
{
var chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
var random = new Random();
var result = new string(Enumerable
.Repeat(chars, 8)
.Select(s => s[random.Next(s.Length)])
.ToArray());
return result;
}
}
消费者：
class Consumer
{
Producer _producer;
string _name;
public Consumer(
Producer producer,
string name)
{
_producer = producer;
_name = name;
}
public string GetOrderedString()
{
string produced = _producer.GetRandomString(_name);
return String.Join(String.Empty, produced.OrderBy(c => c));
}
}
控制台应用程序：
class Program
{
static void Main(string[] args)
{
int consumerNumber = 5;
int reps = 10;
Producer prod = new Producer();
prod.Start();
Task[] consumers = new Task[consumerNumber];
for (var cConsumers = 0; cConsumers < consumerNumber; cConsumers++)
{
Consumer consumer = new Consumer(prod, String.Format("Consumer{0}", cConsumers + 1));
Task consumerTask = Task.Factory.StartNew((consumerIndex) =>
{
int cConsumerNumber = (int)consumerIndex;
for (var counter = 0; counter < reps; counter++)
{
string data = consumer.GetOrderedString();
Console.WriteLine("Consumer{0} consumed {1} at iteration {2}", cConsumerNumber, data, counter + 1);
}
}, cConsumers + 1);
consumers[cConsumers] = consumerTask;
}
Task continuation = Task.Factory.ContinueWhenAll(consumers, (c) =>
{
prod.Dispose();
Console.WriteLine("Producer/Consumer ended");
Console.ReadLine();
});
continuation.Wait();
}
}
我关心的是这是否是解决问题的正确方法，或者你们是否可以建议任何其他最佳实践。
我已经用谷歌搜索并尝试了不同的建议想法，但我尝试过的每个例子都假设生产者能够在他们被要求后立即生产物品……在现实世界的应用程序中这种情况非常罕见:)
非常感谢任何帮助。
我认为信号量并不是真的需要。 Task class and the available concurrent collections.
我试着创建了一些示例代码。不幸的是，我仍然没有完全理解 async/await，所以也许这也是一个对你的情况有帮助的领域（如果你的真正任务主要是 I/O 约束而不是 cpu绑定）。
但是正如您已经看到的，不需要信号量或类似的东西。所有这些事情都是由提供的 类 完成的（例如调用 BlockingCollection.GetConsumingEnumerable()）。
所以希望这能有所帮助。
Program.cs
private static void Main(string[] args)
{
var producer = new Producer();
var consumer = new Consumer(producer.Workers);
consumer.Start();
producer.Start();
Console.ReadKey();
}
Producer.cs
public class Producer : IDisposable
{
private CancellationTokenSource _Cts;
private Random _Random = new Random();
private int _WorkCounter = 0;
private BlockingCollection<Task<String>> _Workers;
private Task _WorkProducer;
public Producer()
{
_Workers = new BlockingCollection<Task<String>>();
}
public IEnumerable<Task<String>> Workers
{
get { return _Workers.GetConsumingEnumerable(); }
}
public void Dispose()
{
Stop();
}
public void Start()
{
if (_Cts != null)
throw new InvalidOperationException("Producer has already been started.");
_Cts = new CancellationTokenSource();
_WorkProducer = Task.Factory.StartNew(() => Run(_Cts.Token));
}
public void Stop()
{
var cts = _Cts;
if (cts != null)
{
cts.Cancel();
cts.Dispose();
_Cts = null;
}
_WorkProducer.Wait();
}
private String GetRandomString()
{
var chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
var result = new String(Enumerable
.Repeat(chars, 8)
.Select(s => s[_Random.Next(s.Length)])
.ToArray());
return result;
}
private void Run(CancellationToken token)
{
while (!token.IsCancellationRequested)
{
var worker = StartNewWorker();
_Workers.Add(worker);
Task.Delay(100);
}
_Workers.CompleteAdding();
_Workers = new BlockingCollection<Task<String>>();
}
private Task<String> StartNewWorker()
{
return Task.Factory.StartNew<String>(Worker);
}
private String Worker()
{
var workerId = Interlocked.Increment(ref _WorkCounter);
var neededTime = TimeSpan.FromSeconds(_Random.NextDouble() * 5);
Console.WriteLine("Worker " + workerId + " starts in " + neededTime);
Task.Delay(neededTime).Wait();
var result = GetRandomString();
Console.WriteLine("Worker " + workerId + " finished with " + result);
return result;
}
}
Consumer.cs
public class Consumer
{
private Task _Consumer;
private IEnumerable<Task<String>> _Workers;
public Consumer(IEnumerable<Task<String>> workers)
{
if (workers == null)
throw new ArgumentNullException("workers");
_Workers = workers;
}
public void Start()
{
var consumer = _Consumer;
if (consumer == null
|| consumer.IsCompleted)
{
_Consumer = Task.Factory.StartNew(Run);
}
}
private void Run()
{
foreach (var worker in _Workers)
{
var result = worker.Result;
Console.WriteLine("Received result " + result);
}
}
}
如果我对你的理解是正确的，你想确保一次只有一个任务被所谓的 "producer" 处理。然后对您的代码稍作修改，您可以这样做：
internal class Producer : IDisposable {
private readonly BlockingCollection<RandomStringRequest> _collection;
public Producer() {
_collection = new BlockingCollection<RandomStringRequest>(new ConcurrentQueue<RandomStringRequest>());
}
public void Start() {
Task consumer = Task.Factory.StartNew(() => {
try {
foreach (var request in _collection.GetConsumingEnumerable()) {
Thread.Sleep(100); // long work
request.SetResult(GetRandomString());
}
}
catch (InvalidOperationException) {
Console.WriteLine("Adding was compeleted!");
}
});
}
public RandomStringRequest GetRandomString(string consumerName) {
var request = new RandomStringRequest();
_collection.Add(request);
return request;
}
public void Dispose() {
_collection.CompleteAdding();
}
private string GetRandomString() {
var chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
var random = new Random();
var result = new string(Enumerable
.Repeat(chars, 8)
.Select(s => s[random.Next(s.Length)])
.ToArray());
return result;
}
}
internal class RandomStringRequest : IDisposable {
private string _result;
private ManualResetEvent _signal;
public RandomStringRequest() {
_signal = new ManualResetEvent(false);
}
public void SetResult(string result) {
_result = result;
_signal.Set();
}
public string GetResult() {
_signal.WaitOne();
return _result;
}
public bool TryGetResult(TimeSpan timeout, out string result) {
result = null;
if (_signal.WaitOne(timeout)) {
result = _result;
return true;
}
return false;
}
public void Dispose() {
_signal.Dispose();
}
}
internal class Consumer {
private Producer _producer;
private string _name;
public Consumer(
Producer producer,
string name) {
_producer = producer;
_name = name;
}
public string GetOrderedString() {
using (var request = _producer.GetRandomString(_name)) {
// wait here for result to be prepared
var produced = request.GetResult();
return String.Join(String.Empty, produced.OrderBy(c => c));
}
}
}
请注意，生产者是单线程的，它使用 GetConsumingEnumerable。也没有信号量，也没有任务。相反，RandomStringRequest 返回给消费者，并且在调用 GetResult 或 TryGetResult 时，它将等待生产者产生结果（或超时到期）。您可能还想在某些地方传递 CancellationTokens（例如 GetConsumingEnumerable）。