Semaphore实现的生产者消费者程序

时间:2023-01-01 19:16:19

Semaphore:Semaphores are often used to restrict the number of threads than can access some (physical or logical) resource.

import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Semaphore;

class Buffer {

List
<Integer> buffer = new LinkedList<Integer>();
// 互斥量,控制buffer的互斥访问
private Semaphore mutex = new Semaphore(1);

// canProduceCount可以生产的数量(表示缓冲区可用的数量)。 通过生产者调用acquire,减少permit数目
private Semaphore canProduceCount = new Semaphore(10);

// canConsumerCount可以消费的数量。通过生产者调用release,增加permit数目
private Semaphore canConsumerCount = new Semaphore(0);
Random rn
= new Random(10);

public void get() throws InterruptedException {
canConsumerCount.acquire();
try {
mutex.acquire();
int val = buffer.remove(0);
System.out
.println(Thread.currentThread().getName()
+ " 正在消费数据为:" + val + " buffer目前大小为:" + buffer.size());
}
finally {
mutex.release();
canProduceCount.release();
}

}

public void put() throws InterruptedException {
canProduceCount.acquire();
try {
mutex.acquire();
int val = rn.nextInt(10);
buffer.add(val);
System.out
.println(Thread.currentThread().getName()
+ " 正在生产数据为:" + val + " buffer目前大小为:" + buffer.size());
}
finally {
mutex.release();
// 生产者调用release,增加可以消费的数量
canConsumerCount.release();
}

}
}

public class SemaphoreProducerComsumer1 {

public static void main(String[] args) {
final Buffer buffer = new Buffer();
startProducer(buffer);
startProducer(buffer);
startConsumer(buffer);
startConsumer(buffer);

}

public static void startProducer(final Buffer buffer) {
new Thread(new Runnable() {

@Override
public void run() {
try {
while (true) {
buffer.put();
}
}
catch (Exception e) {
e.printStackTrace();
}

}
}).start();
}

public static void startConsumer(final Buffer buffer) {
new Thread(new Runnable() {

@Override
public void run() {
try {
while (true) {
buffer.get();
}
}
catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}
}).start();
}

}