C# Producer Consumer （生产者消费者模式）demo

刚接触C#不久，想做个C，C++ Server中经常使用的Producer Consumer模式，以满足多任务排队多线程处理的模式。于是写了下面的demo。

第一套代码将producer Consumer的逻辑写到from类里了，方便在demo的显示界面动态显示模拟生产和消费的过程。

第二套代码将producer Consumer的逻辑单独写到一个类中，使用委托的方法在from中回显生产消费的过程。

Demo中均以3个线程作为消费者，1个线程作为生产者为例。

由于是C#新手，有不对的地方望不吝赐教。

先贴下窗口截图，动态生产消费进度显示



第一套代码：

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Windows.Forms;
using System.Threading;

namespace ProducerConsumerDemo
{
public partial class Form1 : Form
{

public Form1()
{
InitializeComponent();
}

private void button1_Click(object sender, EventArgs e)
{
StopFlag = false;
StartDemo();
}

private void button2_Click(object sender, EventArgs e)
{
StopFlag = true;
}

public class TaskInfo
{
public int Index;

public TaskInfo()
{
}

public TaskInfo(int index)
{
this.Index = index;
}
}

private static bool StopFlag = false;
private int ProducerNumber = 1;
private int ConsumerNumber = 3;

// 共用任务队列
private Queue<TaskInfo> TaskQueue = new Queue<TaskInfo>();
private Semaphore TaskSemaphore = new Semaphore(0, 256);

private object myLock = new object();

private void StartDemo()
{
StartProducer(ProducerNumber);
Thread.Sleep(2000);
StartConsumer(ConsumerNumber);
}

private void StartProducer(int num)
{
for (int i = 0; i < num; i++)
{
Thread t = new Thread(new ThreadStart(Producer));
t.Name = "Producer_" + (i + 1);
t.Start();
}
}

private void StartConsumer(int num)
{
for (int i = 0; i < num; i++)
{
Thread t = new Thread(new ParameterizedThreadStart(this.Consumer));
t.Name = "Consumer_" + (i + 1);
t.IsBackground = true;
t.Start(i + 1);
}
}

// 线程间控件调用，需要使用委托
public delegate void ProducerTextDelegate(string AddStr);
public delegate void ComsumerTextDelegate(int Index, string AddStr);

// 生产者
private void Producer()
{
int i = 0;
while (!StopFlag)
{
TaskInfo Task = new TaskInfo(++i);
lock (myLock)	// 任务队列为临界资源，需要锁
{
TaskQueue.Enqueue(Task);
}
TaskSemaphore.Release(1);	// 每添加一个任务，信号量加1

string AddText = String.Format("Task {0} was Produced.\r\n", Task.Index);
ChangeProducerText(AddText);	// 修改textBox输出

Thread.Sleep(1000);
}
}

// 消费者
private void Consumer(object data)
{
int Index = (int)data;
TaskInfo GetTask = null;

while (true)
{
TaskSemaphore.WaitOne();
lock (myLock)
{
GetTask = TaskQueue.Dequeue();
}

string AddStr = String.Format("Consumer {0} take Task {1}\r\n", Index, GetTask.Index);

ChangeConsumerText(Index, AddStr);
Random r = new Random();
Thread.Sleep(r.Next(5) * 1000);		// 休眠随机事件，模拟实际任务的每个任务的不同时常
// 此处应为任务处理
}
}

private void ChangeProducerText(string AddStr)
{
if (textBox1.InvokeRequired)
{
ProducerTextDelegate pd = new ProducerTextDelegate(ChangeProducerText);
textBox1.Invoke(pd, new object[] { AddStr });
}
else
{
textBox1.AppendText(AddStr);
}
}

private void ChangeConsumerText(int Index, string AddStr)
{
TextBox currentTextBox = null;
switch (Index)
{
case 1:
currentTextBox = this.textBox2;
break;
case 2:
currentTextBox = this.textBox3;
break;
case 3:
currentTextBox = this.textBox4;
break;
default:
break;
}

if (currentTextBox.InvokeRequired)
{
ComsumerTextDelegate cd = new ComsumerTextDelegate(ChangeConsumerText);
currentTextBox.Invoke(cd, new object[] { Index, AddStr });
}
else
{
currentTextBox.AppendText(AddStr);
}
}
}
}

第二套代码：

// Form1.cs
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Windows.Forms;
using System.Threading;

namespace ProducerConsumerDemo
{
// 定义修改输出的委托
public delegate void ShowTextDelegate(int Index, string Addstr);

public partial class Form1 : Form
{
Work work = null;

public Form1()
{
InitializeComponent();
work = new Work();
work.ShowTextEvent += new ShowTextDelegate(ChangeTextValue);
}

private void button1_Click(object sender, EventArgs e)
{
work.Start();
}

private void button2_Click(object sender, EventArgs e)
{
work.Stop();
}

// new Work 中再起的子线程调用的ShowTextDelegate委托事件，所以需要委托才能调用textBox控件
public delegate void ChangeTextDelegate(int Index, string AddStr);

private void ChangeTextValue(int Index, string AddStr)
{
TextBox ChangeTextBox = null;
switch (Index)
{
case 1:
ChangeTextBox = this.textBox1;
break;
case 2:
ChangeTextBox = this.textBox2;
break;
case 3:
ChangeTextBox = this.textBox3;
break;
case 4:
ChangeTextBox = this.textBox4;
break;
default:
break;
}

if (ChangeTextBox.InvokeRequired)
{
ChangeTextDelegate pd = new ChangeTextDelegate(ChangeTextValue);
ChangeTextBox.Invoke(pd, new object[] { Index, AddStr });
}
else
{
ChangeTextBox.AppendText(AddStr);
}

}
}

}

// Work.cs
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace ProducerConsumerDemo
{
public class Work
{
public Work()
{
StopFlag = false;
}

public void Start()
{
StopFlag = false;
StartDemo();
}

public void Stop()
{
StopFlag = true;
}

public class TaskInfo
{
public int Index;

public TaskInfo()
{
}

public TaskInfo(int index)
{
this.Index = index;
}
}

private static bool StopFlag = false;
private int ProducerNumber = 1;
private int ConsumerNumber = 3;

private Queue<TaskInfo> TaskQueue = new Queue<TaskInfo>();
private Semaphore TaskSemaphore = new Semaphore(0, 256);

private object myLock = new object();

public event ShowTextDelegate ShowTextEvent;

private void StartDemo()
{

StartProducer(ProducerNumber);
Thread.Sleep(2000);
StartConsumer(ConsumerNumber);
}

private void StartProducer(int num)
{
for (int i = 0; i < num; i++)
{
Thread t = new Thread(new ThreadStart(Producer));
t.Name = "Producer_" + (i + 1);
t.Start();
}
}

private void StartConsumer(int num)
{
for (int i = 0; i < num; i++)
{
Thread t = new Thread(new ParameterizedThreadStart(this.Consumer));
t.Name = "Consumer_" + (i + 1);
t.IsBackground = true;
t.Start(i + 1);
}
}

private void Producer()
{
int i = 0;
while (!StopFlag)
{
TaskInfo Task = new TaskInfo(++i);
lock (myLock)
{
TaskQueue.Enqueue(Task);
}
TaskSemaphore.Release(1);

string AddText = String.Format("Task {0} was Produced.\r\n", Task.Index);
ShowTextEvent.Invoke(1, AddText);

Thread.Sleep(1000);
}
}

private void Consumer(object data)
{
int Index = (int)data;
TaskInfo GetTask = null;

while (true)
{
TaskSemaphore.WaitOne();
lock (myLock)
{
GetTask = TaskQueue.Dequeue();
}

string AddStr = String.Format("Consumer {0} take Task {1}\r\n", Index, GetTask.Index);
ShowTextEvent.Invoke(Index + 1, AddStr);
Random r = new Random();
Thread.Sleep(r.Next(5) * 1000);
}
}
}
}

有错误的地方请指正，谢谢。