Mutex是互斥体 广泛地应用在多线程编程中 本文以广为流程的Doug Lea的concurrent工具包的Mutex实现为例 进行一点探讨 在Doug Lea的concurrent工具包中 Mutex实现了Sync接口 该接口是concurrent工具包中所有锁（lock） 门（gate）和条件变量（condition）的公共接口 Sync的实现类主要有 Mutex Semaphore及其子类 Latch CountDown ReentrantLock等 这也体现了面向抽象编程的思想 使我们可以在不改变代码或者改变少量代码的情况下 选择使用Sync的不同实现 下面是Sync接口的定义

 public interface Sync{　public void acquire() throws InterruptedException　//获取许可　public boolean attempt(long msecs) throws InterruptedException　//尝试获取许可　public void release()　//释放许可}

    通过使用Sync可以替代Java synchronized关键字 并提供更加灵活的同步控制 当然 并不是说 concurrent工具包是和Java synchronized独立的技术 其实concurrent工具包也是在synchronized的基础上搭建的 从下面对Mutex源码的解析即可以看到这一点 synchronized关键字仅在方法内或者代码块内有效 而使用Sync却可以跨越方法甚至通过在对象之间传递 跨越对象进行同步 这是Sync及concurrent工具包比直接使用synchronized更加强大的地方

    注意Sync中的acquire（）和attempt（）都会抛出InterruptedException 所以使用Sync及其子类时 调用这些方法一定要捕获InterruptedException 而release（）方法并不会抛出InterruptedException 这是因为在acquire（）和attempt（）方法中可能会调用wait（）等待其它线程释放锁 而release（）在实现上进行了简化 直接释放锁 不管是否真的持有 所以 你可以对一个并没有acquire（）的线程调用release（）这也不会有什么问题 而由于release（）不会抛出InterruptedException 所以我们可以在catch或finally子句中调用release（）以保证获得的锁能够被正确释放 比如

  class X{　Sync gate// 　public void m()　{try{　gate acquire()　// block until condition holds　try　{// method body　}　finally { gate release()}}catch (InterruptedException ex) { // evasive action }　}}

(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()

}

}

--------------------------------------------