【杂谈】再看生产-消费模式

public class Lock {
    /**
     * 等待锁的线程队列
     */
    private List<Thread> waitThreads = new ArrayList<>();
    /**
     * 守卫
     */
    private AtomicInteger guard = new AtomicInteger(0);
    /**
     * 锁标志
     */
    private AtomicInteger lockFlag = new AtomicInteger(0);
    /**
     * 当前线程的拥有者
     */
    private Thread holder;

    public void lock() {
        if(Objects.equals(holder, Thread.currentThread())) //如果线程已经获得锁，则直接返回
            return;
        while(!guard.compareAndSet(0, 1)) //尝试获得守卫允许
            ;
        if(lockFlag.intValue() == 0) {
            lockFlag.set(1); //将锁标记为"已被占用"
            holder = Thread.currentThread(); //将锁的拥有者设为当前线程
            guard.set(0); //释放守卫
        } else {
            waitThreads.add(Thread.currentThread()); //加入到等待队列
            guard.set(0); //释放守卫
            LockSupport.park(); //将当前线程挂起
            holder = Thread.currentThread(); //当线程从上一行恢复执行的时候，就说明此线程获得了锁
        }
    }

    public void unlock() {
        if(!Objects.equals(holder, Thread.currentThread())) //如果不是锁的拥有者就没资格释放锁
            return;
        while(!guard.compareAndSet(0, 1))
            ;
        if(waitThreads.size() == 0) { //判断是否有线程正在等待
            lockFlag.set(0); //如果没有，就将锁标记为"空闲"
            holder = null;
            guard.set(0); //释放守卫
        } else {
            LockSupport.unpark(waitThreads.remove(0)); //如果有线程在等待，则唤醒队列中的第一个
            guard.set(0); //释放守卫
        }
    }
}

public class BufferCache {
    /**
     * 缓冲数组，用于保存数据
     */
    private Object[] data;
    /**
     * 读索引 => 下一个要消费的数据从哪里拿
     */
    private int readIndex;
    /**
     * 写索引 => 下一个进来的数据要放哪里
     */
    private int writeIndex;
    /**
     * 当前缓冲区内数据的数量
     */
    private int count;
    /**
     * 生产者线程等待队列
     */
    private List<Thread> waitProducers = new ArrayList<>();
    /**
     * 消费者线程等待队列
     */
    private List<Thread>  waitConsumers = new ArrayList<>();
    /**
     * 前面实现的锁
     */
    private Lock lock = new Lock();

    public BufferCache(int initial) {
        this.data = new Object[initial];
    }

    public void put(Object e) {
        lock.lock(); //获得锁
        while(count == data.length) { //如果已满，就将生产线程挂起
            waitProducers.add(Thread.currentThread()); //放入等待队列，这样要唤醒的时候可以找得到
            lock.unlock(); //释放锁
            LockSupport.park(); //挂起当前线程
            lock.lock(); //醒来的时候，要再获得锁

        }

        data[writeIndex] = e;  //存入数据
        count++;
        if(++writeIndex == data.length) { //循环利用存储空间
            writeIndex = 0;
        }
        while(waitConsumers.size() != 0) { //唤醒消费线程
            LockSupport.unpark(waitConsumers.remove(0));
        }
        lock.unlock(); //释放锁
    }

    public Object take() { //同理
        lock.lock();
        Object e = null;
        while (count == 0) {
            waitConsumers.add(Thread.currentThread());
            lock.unlock();
            LockSupport.park();
            lock.lock();
        }

        e = data[readIndex];
        count--;
        if(++readIndex == data.length) {
            readIndex = 0;
        }
        while(waitProducers.size() != 0) {
            LockSupport.unpark(waitProducers.remove(0));
        }
        lock.unlock();
        return e;
    }

    private static class Task1 implements Runnable { //生产任务
        private int num;
        private BufferCache cache;
        private String name;

        public Task1(BufferCache cache, String index) {
            this.name = "producer-" + index;
            this.num = 0;
            this.cache = cache;
        }

        @Override
        public void run() {
            String data;
            while(true) { //每隔一秒往队列丢一个数据
                data = num + " from " + name;
                cache.put(data);
                System.out.println(name + "放入:" + data);
                num++;
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    private static class Task2 implements Runnable {
        private BufferCache cache;

        public Task2( BufferCache cache) {
            this.cache = cache;
        }

        @Override
        public void run() {
            while(true) { //不断从队列中抓取数据
                Object e = cache.take();
                System.out.println("消费到：" + e);
            }
        }
    }
    public static void main(String[] args) { //跑个案例
        BufferCache cache = new BufferCache(20);
        Thread producer;
        Thread consumer;
        for(int i = 0; i < 5; i++) { //开5个生产者
            producer = new Thread(new Task1(cache, i + ""));
            producer.start();
        }

        for(int i = 0; i < 3; i++) { //开3个消费者
            consumer = new Thread(new Task2(cache));
            consumer.start();
        }
    }
}

Condition x = lock.newCondition();