最近在复习操作系统,随手写两种Java实现生产者消费者模型的方式
一、信号量
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;
class MessageQueue {
private static final Semaphore FULL = new Semaphore(0);
private static final Semaphore EMPTY = new Semaphore(10); // 初始队列为空
private static final Semaphore MUTEX = new Semaphore(1); // 互斥锁
private static final Random RAND_NUM_PRODUCER = new Random(System.currentTimeMillis());
private static final Queue<Integer> QUEUE = new ConcurrentLinkedQueue<>();
public void produce(){
try {
EMPTY.acquire();
MUTEX.acquire();
QUEUE.offer(RAND_NUM_PRODUCER.nextInt(100));
System.out.println("【生产】生产者:" + Thread.currentThread().getName() + " 当前队列:" + QUEUE.size());
} catch (InterruptedException e) {
e.printStackTrace();
}
FULL.release();
MUTEX.release();
}
public void consume(){
try {
FULL.acquire();
MUTEX.acquire();
QUEUE.poll();
System.out.println("【消费】消费者:" + Thread.currentThread().getName() + " 当前队列:" + QUEUE.size());
} catch (InterruptedException e) {
e.printStackTrace();
}
EMPTY.release();
MUTEX.release();
}
}
二、Lock和Condition
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
class MessageQueue {
private static final Lock LOCK = new ReentrantLock();
private static final Condition NOT_FULL = LOCK.newCondition();
private static final Condition NOT_EMPTY = LOCK.newCondition();
private static final int MAX_SIZE = 10;
private static final Random RAND_NUM_PRODUCER = new Random(System.currentTimeMillis());
private static final Queue<Integer> QUEUE = new ConcurrentLinkedQueue<>();
public void produce() {
LOCK.lock();
try{
while(QUEUE.size() == MAX_SIZE){
NOT_FULL.await();
}
QUEUE.offer(RAND_NUM_PRODUCER.nextInt(100));
System.out.println("【生产】生产者:" + Thread.currentThread().getName() + " 当前队列:" + QUEUE.size());
NOT_EMPTY.signalAll();
}catch (InterruptedException e){
e.printStackTrace();
}finally {
LOCK.unlock();
}
}
public void consume(){
LOCK.lock();
try{
while(QUEUE.size() == 0){
NOT_EMPTY.await();
}
QUEUE.poll();
System.out.println("【消费】消费者:" + Thread.currentThread().getName() + " 当前队列:" + QUEUE.size());
NOT_FULL.signalAll();
}catch (InterruptedException e){
e.printStackTrace();
}finally {
LOCK.unlock();
}
}
}
// 以下是测试代码
public class ProducerAndConsumer{
public static void main(String[] args) {
MessageQueue messageQueue = new MessageQueue();
Thread producer = new Thread(()->{
while(true) messageQueue.produce();
});
Thread consumer = new Thread(()->{
while(true) messageQueue.consume();
});
producer.start();
consumer.start();
}
}