食堂中的生产-消费模型

栏目: 数据库 · 发布时间: 6年前

内容简介：在学校的时候，我不爱去食堂成功，一是由于暗黑料理，更重要的一点是人太多了，队伍往往从窗口排到了门口，点菜、计算价格、付款三种业务由打饭阿姨一人完成，思维切换忙碌，操作变更频繁，导致效率低下，降低了食堂的吞吐量，造成了不好的用户体验。而最近在公司食堂吃饭，发现是另外一种设计：工作人员向餐台上添加食物，两条通道对顾客进行分流，顾客依次进入队列然后对餐台上的食物进行自取，最后在队列出口进行结账。变成了这样：

在学校的时候，我不爱去食堂成功，一是由于暗黑料理，更重要的一点是人太多了，队伍往往从窗口排到了门口，点菜、计算价格、付款三种业务由打饭阿姨一人完成，思维切换忙碌，操作变更频繁，导致效率低下，降低了食堂的吞吐量，造成了不好的用户体验。

而最近在公司食堂吃饭，发现是另外一种设计：工作人员向餐台上添加食物，两条通道对顾客进行分流，顾客依次进入队列然后对餐台上的食物进行自取，最后在队列出口进行结账。

变成了这样：

在学校食堂提供的打饭服务中，是由打饭阿姨单独完成一系列的操作。而在公司食堂，则将打饭解耦成了排盘与收银两个服务，降低了每个工作人员的工作复杂度，使其能更加专注于自己负责的工作。同时，在学校食堂中，一名学生在打饭的同时，排在队伍后面的同学是出于等待状态，等待获取打饭阿姨的服务，这对于打饭阿姨的性能是极大的浪费。

而在公司食堂，这是一种生产-消费的模型。我们来看看都有些什么？

首先，消费者是谁？肯定是来用餐的顾客，消费的东西是什么？是餐台上的食物。而生产者则是将食物摆上餐台的工作人员。

那么将这个场景抽象化，可以用这张图来表示：

这张图主要包含了这些信息：

一个数据仓库

用于数据缓存，生产者生产完数据放入，消费者从中取出。它有两个注意点：

同步，即生产者和消费者不能同时访问数据仓库。对应到现实世界，你可以理解为：顾客和工作人员不会在一个瞬间去访问餐台，两个顾客也不能在同一瞬间去争抢同一盘食物。
当仓库满了，生产者无法再添加数据进去，将处于阻塞状态，并提醒消费者进行消费；当仓库为空，消费者无法从中获取数据。处于阻塞状态，并提醒生产者进行生产。

两种角色：生产者、消费者。
两种关系：

依赖关系：生产者与消费者
竞争关系：生产者与生产者、消费者与消费者

那如何使用这种模型，将食堂这个场景来实现还原呢？我们可以这么分析：

数据仓库可以使用一个队列或数组来实现。但是要注意同步。我们这里选择java.util.concurrent包下的BlockingQueue，它有着已经实现的阻塞添加和阻塞删除的特性。当然也可以用一个List去自己实现。
生产者是一些线程，生产一些数据，将其入队。
消费者是一些线程，从队首获取数据，将其消费。

核心概念就是以上三点，那么我们接下来用代码简单实现一下。预期目标是让整个流程自动运行起来。

首先我们来定义一个食物类 Food ，这里从简，只给它一个 id 属性。

public class Food {
    private int id;
    //含参构造函数，这里用不到无参构造器，所以可以不写。否则新建对象将会出错。
    public Food(int id){
        this.id=id;
    }
    //get/set
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
}
复制代码

接着我们来编写生产者类 Producer 和消费者 Consumer ,让我们来想想这两个类该怎么去编写？

首先我们需要使用组合的方式来加入一条 BlockingQueue （阻塞队列），它是整个问题的核心，又由于这个容器是只用来存放 Food 的，我们可以加上泛型。

之前也提到了，Food有一个属性是id，那么这个id从何而来呢？应该是生产者生产Food时赋予的，而由于多位生产者在同时生产，为了保证id的唯一性，我们需要对其进行原子化的自增操作。这里我们可以使用java.util.concurrent.atomic包中的AtomicInteger，当然如果使用int，然后自己做一些同步操作也是可以的。（若这里对 Java 内存模型JMM不了解的，可以去阅读我写过的这篇文章：）

那么， Producer 类可以这么写,将每个工作线程当作一名生产者，并作一些必要的文字输出，方便控制台观察:

public class Producer implements Runnable {
    private boolean working=true;
    private BlockingQueue<Food> queue;
    private static AtomicInteger count = new AtomicInteger();
    //构造函数
    public Producer(BlockingQueue<Food> queue){
        this.queue=queue;
    }
    @Override
    public void run() {
        while(working){
            int id=count.incrementAndGet();
            Food food=new Food(id);
//            System.out.println(Thread.currentThread().getId()+"号员工开始工作");
            if(queue.offer(food)){
                System.out.println(Thread.currentThread().getId()+"号员工将"+food.getId()+"号食物加入餐台");
            }else {
                System.out.println("餐台已满，"+food.getId()+"号食物无法加入");
            }
            try {
                Thread.sleep(1000*3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public  void stop(){
        working=false;
    }
}
复制代码

同理， Consumer 可以这么写：

public class Consumer implements Runnable {
    private boolean working=true;
    private BlockingQueue<Food> queue;
//含参构造函数
    public Consumer(BlockingQueue<Food> queue){
        this.queue=queue;
    }
    @Override
    public void run() {
        while (true){
            try {
                Food food=queue.take();//take()方式，若队列中没有元素则线程被阻塞
                System.out.println(food.getId()+"号食物已被"+Thread.currentThread().getId()+"号顾客端走");
                Thread.sleep(1000*2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}
复制代码

最后，就是主函数调用了，我们使用线程池来对线程进行分配，这里我将数据队列定义为10个元素空间，线程池使用了newFixedThreadPool方式来规定5条线程，初始化3名生产者和15名消费者。

Main.java

public class Main {
    public static void main(String[] args) {
        BlockingQueue<Food> queue = new LinkedBlockingDeque<>(10);
        Producer[] p=new Producer[3];
        Consumer[] c=new Consumer[15];
        for (int i=0;i<3;i++){
            p[i]=new Producer(queue);
        }
        for (int j=0;j<15;j++){
            c[j]=new Consumer(queue);
        }
        ExecutorService executorService= Executors.newFixedThreadPool(5);
        for (int i=0;i<3;i++){
            executorService.execute(p[i]);
        }
        for (int j=0;j<15;j++){
            executorService.execute(c[j]);
        }
        try {
            Thread.sleep(1000*20);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
复制代码

下面就是食堂开张的时刻了:stuck_out_tongue_closed_eyes: 让我们赶紧运行试试。