生产者和消费者手撕代码

实在智能二面的时候面试官让手撕这个代码:

某快递仓库最多存放10件包裹。货车司机负责送货，快递员负责取货：      
1、货车每次运输1件包裹到仓库，耗时0.3秒，累计运输5次后停止
2、快递员每次取1件包裹，耗时0.5秒，当仓库空且货车停止时结束工作
3、每次存取需实时输出余量，如："货车存入1件，当前库存：3/10"

快递员可以认为是消费者，货车是生产者不断往里面去生产内容。

仓库就是缓冲区，用来存储物品，这就是一个典型的生产者和消费者的模型结构。

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;

public class Task {

    //模拟缓冲区
    private static LinkedBlockingDeque<Integer> queue = new LinkedBlockingDeque<>();
    //耗时0.3秒
    private static Integer DriverTime = 300;
    private static Integer workerTime = 500;

    //数量  使用原子类保证了并发情况下面的一致性
    private static AtomicInteger value = new AtomicInteger(0);
    //最大数量 
    private static Integer maxValue = 10;
    private static Integer count = 5;


    private static boolean isWorked = false;

    public static void main(String[] args) {
        Thread driver1 = new Thread(new Driver());
        Thread driver2 = new Thread(new Driver());
        Thread worker = new Thread(new worker());
        worker.start();
        driver1.start();
        driver2.start();
    }

    static class Driver implements Runnable {

        @Override
        public void run() {
            try {
                //耗时0.3s
                Thread.sleep(DriverTime);
                //五次
                for (int i = 0; i < count; i++) {
                    //物品个数++
                    int currentValue = value.incrementAndGet();
                    //放入到缓冲区中
                    queue.put(currentValue);
                    System.out.println("货车存入1件,当前库存:" + currentValue + "/" + maxValue);
                }
                isWorked = true;
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    static class worker implements Runnable {

        @Override
        public void run() {
            //去除包裹
            //结束的逻辑
            while (true) {
                if (queue.isEmpty() && isWorked == true) {
                    //结束工作
                    break;
                }
                try {
                    Thread.sleep(workerTime);
                    //从缓冲区中取出进行消费
                    Integer getValue = queue.take();
                    System.out.println("当前库存:" + getValue + "/" + maxValue);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
}

面试的时候写了这一版的代码，但是面试官要求有没有更新的，能不能结合类和设计模式进行改造。

后面让 ai 改造了一下，利用一些设计模式和接口等，实现了业务的解耦。

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;

// TaskManager 负责协调司机和工人之间的任务交付
public class TaskManager {
    private static LinkedBlockingDeque<Integer> queue = new LinkedBlockingDeque<>();
    private static final int DRIVER_TIME = 300;
    private static final int WORKER_TIME = 500;
    private static final int MAX_VALUE = 10;
    private static final int COUNT = 5;
    private static AtomicInteger value = new AtomicInteger(0);
    private static boolean isWorked = false;

    public static void main(String[] args) {
        Producer driver1 = new Driver();
        Producer driver2 = new Driver();
        Consumer worker = new Worker();
        
        TaskManager taskManager = new TaskManager();
        
        // 启动司机线程
        new Thread(() -> driver1.produce(taskManager)).start();
        new Thread(() -> driver2.produce(taskManager)).start();
        
        // 启动工人线程
        new Thread(() -> worker.consume(taskManager)).start();
    }

    // 提供线程安全的任务添加方法
    public synchronized void addTask(int task) {
        try {
            queue.put(task);
            System.out.println("货车存入1件,当前库存:" + task + "/" + MAX_VALUE);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    // 提供线程安全的任务获取方法
    public synchronized Integer getTask() {
        try {
            if (queue.isEmpty() && isWorked) {
                return null;
            }
            Integer task = queue.take();
            System.out.println("工人处理1件, 当前库存:" + task + "/" + MAX_VALUE);
            return task;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public static void markWorkCompleted() {
        isWorked = true;
    }

    // 获取当前任务计数
    public AtomicInteger getValue() {
        return value;
    }

    // 获取工作状态
    public boolean isWorked() {
        return isWorked;
    }
}

// Producer接口定义了生产行为
interface Producer {
    void produce(TaskManager taskManager);
}

// Consumer接口定义了消费行为
interface Consumer {
    void consume(TaskManager taskManager);
}

// Driver类实现了Producer接口，负责生产任务
class Driver implements Producer {

    @Override
    public void produce(TaskManager taskManager) {
        try {
            Thread.sleep(TaskManager.DRIVER_TIME);
            for (int i = 0; i < TaskManager.COUNT; i++) {
                int currentValue = taskManager.getValue().incrementAndGet();
                taskManager.addTask(currentValue);
            }
            TaskManager.markWorkCompleted();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

// Worker类实现了Consumer接口，负责消费任务
class Worker implements Consumer {

    @Override
    public void consume(TaskManager taskManager) {
        while (true) {
            Integer task = taskManager.getTask();
            if (task == null) {
                break;
            }
            try {
                Thread.sleep(TaskManager.WORKER_TIME);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}