Java基于ReentrantLock类和Condition接口实现生产者消费者模型

偏执的太偏执、 2023-01-07 10:18 132阅读 0赞

测试用生产者模型：

package domain;
    
    import java.util.Queue;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * 生产者模型
     * 
     * @author WanAkiko
     * 
     */
    public class Producer implements Runnable {
    
    	private Queue<String> queue;
    	private ReentrantLock lock;
    	private Condition condition;
    	private Integer count = 0;
    	int i = 1;
    
    	public Producer(Queue<String> queue, ReentrantLock lock,
    			Condition condition, Integer count) {
    		super();
    		this.queue = queue;
    		this.lock = lock;
    		this.condition = condition;
    		this.count = count;
    	}
    
    	@Override
    	public void run() {
    		while (true) {
    			lock.lock();
    			while (queue.size() >= count) {
    				try {
    					condition.await(); // 线程通信或中断之前保持等待状态
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    			try {
    				Thread.sleep(1000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    			String msg = "生产序列-" + (i++);
    			queue.add(msg);
    			System.out.println("生产者线程：" + msg);
    			condition.signal();
    			lock.unlock();
    		}
    	}
    
    }

测试用消费者模型：

package domain;
    
    import java.util.Queue;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * 消费者模型
     * 
     * @author WanAkiko
     * 
     */
    public class Consumer implements Runnable {
    
    	private Queue<String> queue; // 消费者队列
    	@SuppressWarnings("unused")
    	private Integer count; // 消费数量
    	private ReentrantLock lock; // 唯一对象锁
    	private Condition condition; // 监视器
    
    	public Consumer(Queue<String> queue, Integer count, ReentrantLock lock,
    			Condition condition) {
    		super();
    		this.queue = queue;
    		this.count = count;
    		this.lock = lock;
    		this.condition = condition;
    	}
    
    	@Override
    	public void run() {
    		while (true) {
    			lock.lock(); // 获取互斥锁
    			while (queue.isEmpty()) {
    				try {
    					condition.await(); // 线程通信或中断之前保持等待状态
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    			try {
    				Thread.sleep(1000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    			condition.signal();
    			System.out.println("消费者线程：" + queue.remove());
    			lock.unlock(); // 释放互斥锁
    		}
    	}
    }

模型测试：

package client;
    
    import java.util.LinkedList;
    import java.util.Queue;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.ReentrantLock;
    
    import domain.Consumer;
    import domain.Producer;
    
    public class PC02Test {
    
    	public static void main(String[] args) {
    		
    		Queue<String> queue = new LinkedList<String>();
    		ReentrantLock lock = new ReentrantLock();
    		Condition condition = lock.newCondition();
    		int count = 3;
    		
    		new Thread(new Producer(queue, lock, condition, count)).start();
    		new Thread(new Consumer(queue, count, lock, condition)).start();
    		
    	}
    	
    }

运行示例：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTY1Mzkz_size_16_color_FFFFFF_t_70]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTY1Mzkz_size_16_color_FFFFFF_t_70]: /images/20221119/4d7a3169cae54febbcc8d9c999774874.png