Java多线程——消费者模型

水深无声 2024-03-27 16:27 69阅读 0赞

### 一、模型概述 ###

在一个流水线任务过程中包含了任务产生和任务执行的线程，但较快的线程执行速度会让线程无法按照规定的顺序执行。比如在任务列表为空的情况下，任务执行线程比任务产生线程先一步执行，这样会导致错误的运行结果。因此我们需要在线程中引入控制信号来控制线程执行的先后顺序。而消费者模型就能够有效解决这个问题。

在消费者模型中，生产者会不断产生任务并保存到列表中。同时在生产者线程中也有控制线程状态的功能。消费者即将任务列表中的任务取出并执行。在运行过程中，如果任务列表中的任务已执行完，那么生产者将会控制消费者线程停止，并且向列表中存入一定数量的任务，当任务数量达到某一阈值时生产者线程会唤醒消费者线程继续执行。

### 二、解决方案：管程法 ###

管程法大致原理如下：创建一个共享缓冲区供生产者和消费者使用，生产者向缓冲区中存入任务，消费者向缓冲区中取出任务执行。接着给缓冲区设置一个阈值，当缓冲区中的任务数量小于阈值时，消费线程停止，生产线程启动并向缓冲区中添加任务。当任务数量高于阈值时生产线程停止，消费线程开始向缓冲区中取出任务执行。

代码实现：

（1）、缓冲区类创建

public class BlockingQueue <E>{
    
        //创建一个链表对象用于存储任务的数据对象
        public Queue<E> que = new LinkedList<E>();
        
        //设置链表触发阈值
        int limit;
    
        //初始化对象，并设置阈值大小
        public BlockingQueue(int limit){
            this.limit = limit;
        }
    
        //生产者存入
        public synchronized void push(E data) throws InterruptedException{
            //实时判断表中任务是否达到阈值
            while (que.size() == limit){
                System.out.println("t1.sleep");
                //达到阈值时生产者线程暂停
                wait();
            }
            //当任务列表中对象为空时
            if(que.isEmpty()){
                System.out.println("t1.continue");
                //唤醒生产者线程产生任务
                notifyAll();
            }
            //存入任务
            que.add(data);
        }
    
        //消费者取出
        public synchronized E pop(){
            //实时判断链表是否为空
            while (que.isEmpty()){
                try {
                    System.out.println("t2.sleep");
                    //为空时暂停消费者线程
                    wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            //链表中有任务时
            if(que.size() == limit){
                System.out.println("t2.continue");
                //唤醒消费者线程
                notifyAll();
            }
            //取出任务
            return que.remove();
        }
        
        //链表大小
        public int getSize(){
            return que.size();
        }
    }

（2）、主程序

public class Manage {
        public static void main(String[] args) throws InterruptedException {
            BlockingQueue blockingQueue = new BlockingQueue(50);
    
            //生产者线程
            Thread t1 = new Thread(){
                @Override
                public void run() {
                   //创建死循环
                   for(int i = 0 ; ; i ++){
                       try {
                           Thread.sleep(200);
                       } catch (InterruptedException e) {
                           throw new RuntimeException(e);
                       }
                       try {
                            //将循环的数存入链表中
                           blockingQueue.push(i);
                       } catch (InterruptedException e) {
                           throw new RuntimeException(e);
                       }
                   }
                }
            };
    
            //消费者线程
            Thread t2 = new Thread(){
                @Override
                public void run() {
                    //创建死循环
                    while(true){
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        //取出任务
                        System.out.println(blockingQueue.pop());
                    }
                }
            };
            t1.start();
            Thread.sleep(5000);
            t2.start();
        }
    }

运行结果（节选）：

> 生产者写入  
> 生产者写入  
> 生产者写入  
> 消费者取出  
> 0  
> 消费者取出  
> 1  
> 生产者写入  
> 消费者取出  
> 2  
> 消费者取出  
> 3  
> 生产者写入  
> 消费者取出  
> 4  
> 消费者取出  
> 5  
> 生产者写入  
> 消费者取出  
> 6