下面是我在网上转下来的，看了几篇文章，感觉这篇写得比较好，给你贴上，我有时间也好好消化消化。
--------------------------------------------
（1） Consumer表示消费类，其中定义了一个ArrayList表示容器，容器中装的是Goods这个类的对象。主要是方法Consumption，这个是表示消费的方法。具体代码如下：
public class Consumer
{
ArrayList container = null
//得到一个容器
public Consumer(ArrayList container)
{
this.container = container
}
//定义一个消费的方法
public void Consumption()
{
Goods goods = (Goods)this.container[0]
Console.WriteLine("消费了物品：" + goods.ToString())
//消费掉容器中的一个物品
this.container.RemoveAt(0)
}
}
（2） Producer表示生产类，同样有一个ArrayList表示容器，和Consumer类一样有一个得到容器的构造函数。主要方法是Product，表示生产一个物品的方法，具体代码如下：
public class Producer
{
ArrayList container = null
//得到一个容器
public Producer(ArrayList container)
{
this.container = container
}
//定义一个生产物品的方法装入容器
public void Product(string name)
{
//创建一个新物品装入容器
Goods goods = new Goods()
goods.Name = name
this.container.Add(goods)
Console.WriteLine("生产了物品：" + goods.ToString())
}
}
（3） 接下来看看物品类Goods，这个表示生产者和消费者进行交互的物品，其中就一个属性Name表示名称，同时重写了ToString()方法。具体代码如下：
public class Goods
{
//物品名称
private string name
public string Name
{
get { return name}
set { name = value}
}
//重写ToString()
public override string ToString()
{
return "物品名称：" + name
}
}
（4） 最后看看我们的Programe类，这个类包含了一个启动的Main函数，其中还有两个方法ThreadConsumption和ThreadProduct。这两个方法是放在两个线程中进行执行的。表示一个线程进行生产，一个线程进行消费。生产线程在生产好一个物品后就通知消费者进行物品的消费，自己便停下等消费者消费完毕后通知生产线程再进行生产。具体代码如下：
class Program
{
//创建一个消费者和生产者共用的一个容器
ArrayList container = new ArrayList()
Producer producer = null
Consumer consumer = null
static void Main(string[] args)
{
Program p = new Program()
//创建两个线程并启动
Thread t1 = new Thread(new ThreadStart(p.ThreadProduct))
Thread t2 = new Thread(new ThreadStart(p.ThreadConsumption))
t1.Start()
t2.Start()
Console.Read()
}
//定义一个线程方法生产8个物品
public void ThreadProduct()
{
//创建一个生产者
producer = new Producer(this.container)
lock (this)
{
for (int i = 1i <= 8i++)
{
//如果容器中没有就进行生产
if (this.container.Count == 0)
{
//调用方法进行生产
producer.Product(i + "")
//生产好了一个通知消费者消费
Monitor.Pulse(this)
}
//容器中还有物品等待消费者消费后再生产
Monitor.Wait(this)
}
}
}
//定义一个线程方法消费生产的物品
public void ThreadConsumption()
{
//创建一个消费者
consumer = new Consumer(this.container)
lock (this)
{
while (true)
{
//如果容器中有商品就进行消费
if (this.container.Count != 0)
{
//调用方法进行消费
consumer.Consumption()
Monitor.Pulse(this)
}
//容器中没有商品通知消费者消费
Monitor.Wait(this)
}
}
}
}
使用的生产者和消费者模型具有如下特点：(1)本实验的多个缓冲区不是环形循环的，也不要求按顺序访问。生产者可以把产品放到目前某一个空缓冲区中。
(2)消费者只消费指定生产者的产品。
(3)在测试用例文件中指定了所有的生产和消费的需求，只有当共享缓冲区的数据满足了所有关于它的消费需求后，此共享缓冲区才可以作为空闲空间允许新的生产者使用。
(4)本实验在为生产者分配缓冲区时各生产者间必须互斥，此后各个生产者的具体生产活动可以并发。而消费者之间只有在对同一产品进行消费时才需要互斥，同时它们在消费过程结束时需要判断该消费对象是否已经消费完毕并清除该产品。
Windows
用来实现同步和互斥的实体。在Windows
中，常见的同步对象有：信号量(Semaphore)、
互斥量(Mutex)、临界段(CriticalSection)和事件(Event)等。本程序中用到了前三个。使用这些对象都分
为三个步骤，一是创建或者初始化：接着请求该同步对象，随即进入临界区，这一步对应于互斥量的
上锁；最后释放该同步对象，这对应于互斥量的解锁。这些同步对象在一个线程中创建，在其他线程
中都可以使用，从而实现同步互斥。当然，在进程间使用这些同步对象实现同步的方法是类似的。
1．用锁 *** 作原语实现互斥
为解决进程互斥进人临界区的问题，可为每类临界区设置一把锁，该锁有打开和关闭两种状态，进程执行临界区程序的 *** 作按下列步骤进行：
①关锁。先检查锁的状态，如为关闭状态，则等待其打开；如已打开了，则将其关闭，继续执行步骤②的 *** 作。
②执行临界区程序。
③开锁。将锁打开，退出临界区。
2．信号量及WAIT，SIGNAL *** 作原语
信号量的初值可以由系统根据资源情况和使用需要来确定。在初始条件下信号量的指针项可以置为0，表示队列为空。信号量在使用过程中它的值是可变的，但只能由WAIT，SIGNAL *** 作来改变。设信号量为S，对S的WAIT *** 作记为WAIT（S），对它的SIGNAL *** 作记为SIGNAL（S）。
WAIT（S）：顺序执行以下两个动作：
①信号量的值减1，即S=S－1；
②如果S≥0，则该进程继续执行；
如果
S（0，则把该进程的状态置为阻塞态，把相应的WAITCB连人该信号量队列的末尾，并放弃处理机，进行等待（直至其它进程在S上执行SIGNAL *** 作，把它释放出来为止）。
SIGNAL（S）：顺序执行以下两个动作
①S值加
1，即
S=S＋1；
②如果S）0，则该进程继续运行；
如果S（0则释放信号量队列上的第一个PCB（既信号量指针项所指向的PCB）所对应的进程（把阻塞态改为就绪态），执行SIGNAL *** 作的进程继续运行。
在具体实现时注意，WAIT，SIGNAL *** 作都应作为一个整体实施，不允许分割或相互穿插执行。也就是说，WAIT，SIGNAL *** 作各自都好像对应一条指令，需要不间断地做下去，否则会造成混乱。
从物理概念上讲，信号量S）时，S值表示可用资源的数量。执行一次WAIT *** 作意味着请求分配一个单位资源，因此S值减1；当S<0时，表示已无可用资源，请求者必须等待别的进程释放了该类资源，它才能运行下去。所以它要排队。而执行一次SIGNAL *** 作意味着释放一个单位资源，因此S值加1；若S（0时，表示有某些进程正在等待该资源，因而要把队列头上的进程唤醒，释放资源的进程总是可以运行下去的。
---------------
/**
*
生产者
*
*/
public
class
Producer
implements
Runnable{
private
Semaphore
mutex,full,empty
private
Buffer
buf
String
name
public
Producer(String
name,Semaphore
mutex,Semaphore
full,Semaphore
empty,Buffer
buf){
this.mutex
=
mutex
this.full
=
full
this.empty
=
empty
this.buf
=
buf
this.name
=
name
}
public
void
run(){
while(true){
empty.p()
mutex.p()
System.out.println(name+"
inserts
a
new
product
into
"+buf.nextEmptyIndex)
buf.nextEmptyIndex
=
(buf.nextEmptyIndex+1)%buf.size
mutex.v()
full.v()
try
{
Thread.sleep(1000)
}
catch
(InterruptedException
e)
{
e.printStackTrace()
}
}
}
}
---------------
/**
*
消费者
*
*/
public
class
Customer
implements
Runnable{
private
Semaphore
mutex,full,empty
private
Buffer
buf
String
name
public
Customer(String
name,Semaphore
mutex,Semaphore
full,Semaphore
empty,Buffer
buf){
this.mutex
=
mutex
this.full
=
full
this.empty
=
empty
this.buf
=
buf
this.name
=
name
}
public
void
run(){
while(true){
full.p()
mutex.p()
System.out.println(name+"
gets
a
product
from
"+buf.nextFullIndex)
buf.nextFullIndex
=
(buf.nextFullIndex+1)%buf.size
mutex.v()
empty.v()
try
{
Thread.sleep(1000)
}
catch
(InterruptedException
e)
{
e.printStackTrace()
}
}
}
}
-------------------------
/**
*
缓冲区
*
*/
public
class
Buffer{
public
Buffer(int
size,int
nextEmpty,int
nextFull){
this.nextEmptyIndex
=
nextEmpty
this.nextFullIndex
=
nextFull
this.size
=
size
}
public
int
size
public
int
nextEmptyIndex
public
int
nextFullIndex
}
-----------------
/**
*
此类用来模拟信号量
*
*/
public
class
Semaphore{
private
int
semValue
public
Semaphore(int
semValue){
this.semValue
=
semValue
}
public
synchronized
void
p(){
semValue--
if(semValue<0){
try
{
this.wait()
}
catch
(InterruptedException
e)
{
e.printStackTrace()
}
}
}
public
synchronized
void
v(){
semValue++
if(semValue<=0){
this.notify()
}
}
}
------------------------
public
class
Test
extends
Thread
{
public
static
void
main(String[]
args)
{
Buffer
bf=new
Buffer(10,0,0)
Semaphore
mutex=new
Semaphore(1)
Semaphore
full=new
Semaphore(0)
Semaphore
empty=new
Semaphore(10)
//new
Thread(new
Producer("p001",mutex,full,empty,bf)).start()
Producer
p=new
Producer("p001",mutex,full,empty,bf)
new
Thread(new
Producer("p002",mutex,full,empty,bf)).start()
new
Thread(new
Producer("p003",mutex,full,empty,bf)).start()
new
Thread(new
Producer("p004",mutex,full,empty,bf)).start()
new
Thread(new
Producer("p005",mutex,full,empty,bf)).start()
try{
sleep(3000)
}
catch(Exception
ex)
{
ex.printStackTrace()
}
new
Thread(new
Customer("c001",mutex,full,empty,bf)).start()
new
Thread(new
Customer("c002",mutex,full,empty,bf)).start()
new
Thread(new
Customer("c003",mutex,full,empty,bf)).start()
new
Thread(new
Customer("c004",mutex,full,empty,bf)).start()
new
Thread(new
Customer("c005",mutex,full,empty,bf)).start()
}
}
--------------------------------------------
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）