在 C# 中实现生产者/消费者模式 [英] Implementing the Producer/Consumer Pattern in C#
问题描述
如何使用事件和委托在 C# 中实现生产者/消费者模式?使用这些设计模式时,在资源方面我需要注意什么?是否有任何我需要注意的边缘情况?
How can I implement the Producer/Consumer patterns in C# using Events and Delegates? What do I need to keep an eye out for when it comes to resources when using these design patterns? Are there any edge cases I need to be aware of?
推荐答案
我知道这个帖子有点老了,但是因为我有时在搜索中遇到它,所以我决定分享这个生产者-消费者代码给想知道的人如何实现一个简单的通用生产者-消费者作业队列.
I know this thread is quite a bit old, but since I came across it sometimes in my searches, I decided to share this producer-consumer code for people wondering how to implement a simple generic producer-consumer job queue.
Job 类用于以委托的形式存储"对象的方法调用.然后在处理作业时调用委托.任何相关参数也存储在这个 Job 类中.
The Job class is used to 'store' an object's method call in the form of a delegate. The delegate is then called when the job is processed. Any relevant arguments are also stored in this Job class.
通过这种简单的模式,可以在入队和出队过程中实现多线程.实际上这只是最简单的部分:多线程给您的代码带来了新的挑战,您稍后会注意到它们;-)
With this simple pattern it's possible to achieve multi-threading in the enqueue AND dequeue processes. Actually this is just the easiest part: multi-threading brings new challenges to your code, you'll notice them later ;-)
我最初将此代码发布在此 线程.
I've originally posted this code in this thread.
using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Threading;
// Compiled and tested in: Visual Studio 2017, DotNET 4.6.1
namespace MyNamespace
{
public class Program
{
public static void Main(string[] args)
{
MyApplication app = new MyApplication();
app.Run();
}
}
public class MyApplication
{
private BlockingCollection<Job> JobQueue = new BlockingCollection<Job>();
private CancellationTokenSource JobCancellationTokenSource = new CancellationTokenSource();
private CancellationToken JobCancellationToken;
private Timer Timer;
private Thread UserInputThread;
public void Run()
{
// Give a name to the main thread:
Thread.CurrentThread.Name = "Main";
// Fires a Timer thread:
Timer = new Timer(new TimerCallback(TimerCallback), null, 1000, 2000);
// Fires a thread to read user inputs:
UserInputThread = new Thread(new ThreadStart(ReadUserInputs))
{
Name = "UserInputs",
IsBackground = true
};
UserInputThread.Start();
// Prepares a token to cancel the job queue:
JobCancellationToken = JobCancellationTokenSource.Token;
// Start processing jobs:
ProcessJobs();
// Clean up:
JobQueue.Dispose();
Timer.Dispose();
UserInputThread.Abort();
Console.WriteLine("Done.");
}
private void ProcessJobs()
{
try
{
// Checks if the blocking collection is still up for dequeueing:
while (!JobQueue.IsCompleted)
{
// The following line blocks the thread until a job is available or throws an exception in case the token is cancelled:
JobQueue.Take(JobCancellationToken).Run();
}
}
catch { }
}
private void ReadUserInputs()
{
// User input thread is running here.
ConsoleKey key = ConsoleKey.Enter;
// Reads user inputs and queue them for processing until the escape key is pressed:
while ((key = Console.ReadKey(true).Key) != ConsoleKey.Escape)
{
Job userInputJob = new Job("UserInput", this, new Action<ConsoleKey>(ProcessUserInputs), key);
JobQueue.Add(userInputJob);
}
// Stops processing the JobQueue:
JobCancellationTokenSource.Cancel();
}
private void ProcessUserInputs(ConsoleKey key)
{
// Main thread is running here.
Console.WriteLine($"You just typed '{key}'. (Thread: {Thread.CurrentThread.Name})");
}
private void TimerCallback(object param)
{
// Timer thread is running here.
Job job = new Job("TimerJob", this, new Action<string>(ProcessTimer), "A job from timer callback was processed.");
JobQueue.TryAdd(job); // Just enqueues the job for later processing
}
private void ProcessTimer(string message)
{
// Main thread is running here.
Console.WriteLine($"{message} (Thread: {Thread.CurrentThread.Name})");
}
}
/// <summary>
/// The Job class wraps an object's method call, with or without arguments. This method is called later, during the Job execution.
/// </summary>
public class Job
{
public string Name { get; }
private object TargetObject;
private Delegate TargetMethod;
private object[] Arguments;
public Job(string name, object obj, Delegate method, params object[] args)
{
Name = name;
TargetObject = obj;
TargetMethod = method;
Arguments = args;
}
public void Run()
{
try
{
TargetMethod.Method.Invoke(TargetObject, Arguments);
}
catch(Exception ex)
{
Debug.WriteLine($"Unexpected error running job '{Name}': {ex}");
}
}
}
}
这篇关于在 C# 中实现生产者/消费者模式的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!