synchronized、lock、semaphore、blockingqueue实现生产消费场景。

性能：

数据量少时，Synchronized> Lock、Semaphore。

数据量大时，Lock > Synchronized > Semaphore。

Blockingqueue底层也是使用ReentrantLock + Condition。

一、Synchronized方式

synchronized作用于方法或者对象上，保证方法或者对象在某一时刻只能被一个线程占用。配合wait()、notify()进行线程间通讯。当库存不足时，消费者线程暂停等待。当生产者生产出新的商品时，通知消费者进行消费。

package com.boot.demo;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author braska
 * @date 2020/3/9
 **/
public class SynchronizedTest {

    private AtomicInteger stock = new AtomicInteger(0);

    public synchronized void produce() {
        if (stock.get() < 5) {
            System.out.println(String.format("目前库存：%s", stock.addAndGet(1)));
            notify();
        } else {
            try {
                System.out.println("仓库已满。");
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public synchronized void consume(String name) {
        if (stock.get() > 0) {
            System.out.println(String.format("%s在消费， 目前库存：%s", name,  stock.addAndGet(-1)));
            notify();
        } else {
            try {
                System.out.println("库存已光。");
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class Producer implements Runnable {
        private SynchronizedTest test;

        public Producer(SynchronizedTest test) {
            this.test = test;
        }

        @Override
        public void run() {
            while(true) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                test.produce();
            }
        }
    }

    static class Consumer implements Runnable {
        private SynchronizedTest test;
        private String name;

        public Consumer(SynchronizedTest test, String name) {
            this.test = test;
            this.name = name;
        }

        @Override
        public void run() {
            while(true) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                test.consume(name);
            }
        }
    }


    public static void main(String[] args) {
        SynchronizedTest test = new SynchronizedTest();

        Thread p1 = new Thread(new Producer(test));
        Thread c1 = new Thread(new Consumer(test, "消费者1"));
        Thread c2 = new Thread(new Consumer(test, "消费者2"));

        p1.start();
        c1.start();
        c2.start();
    }
}

二、Semaphore方式

简单点说，Semaphore构造函数定义开放的购物窗口比如new Semaphore(6)，无参默认开放一个窗口。tryAcquire()表示消费者抢占购物窗口，当然一个消费者可以抢占多个窗口比如tryAcquire(2)。当窗口被占满时，后面的消费者需要等待前面的消费者消费完，然后离开(释放)release(2)。

package com.boot.demo;

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

/**
 * @author braska
 * @date 2020/3/9
 **/
public class SemaphoreTest {

    Semaphore semaphore = new Semaphore(1);
    AtomicInteger stock = new AtomicInteger(0);

    public void produce() {
        if (stock.get() < 5) {
            System.out.println(String.format("目前库存：%s", stock.addAndGet(1)));
            semaphore.release();
        } else {
            System.out.println("仓库已满。");
        }
    }

    public void comsume(String name) {
        if (semaphore.tryAcquire()) {
            if (stock.get() > 0) {
                System.out.println(String.format("%s正在消费，目前库存: %s", name, stock.addAndGet(-1)));
            } else {
                System.out.println("库存已光。");
            }
        }
    }

    static class Producer implements Runnable{

        private SemaphoreTest deal;
        public Producer(SemaphoreTest deal) {
            this.deal = deal;
        }

        @Override
        public void run() {
            while (true)
            try {
                Thread.sleep(1000);
                deal.produce();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
    }

    static class Consumer implements Runnable {
        private SemaphoreTest deal;
        private String name;
        public Consumer(SemaphoreTest deal, String name) {
            this.deal = deal;
            this.name = name;
        }

        @Override
        public void run() {
            while (true)
            try {
                Thread.sleep(3000);
                deal.comsume(name);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


    public static void main(String[] args) {
        SemaphoreTest deal = new SemaphoreTest();

        Thread p = new Thread(new Producer(deal));

        Thread c1 = new Thread(new Consumer(deal, "消费者1"));
        Thread c2 = new Thread(new Consumer(deal, "消费者2"));

        p.start();
        c1.start();
        c2.start();
    }
}

三、Lock方式

说起来同synchronized方式是一样的。锁住某一个对象或者执行体保证在某一时刻不被多个线程同时占用。区别：synchronized不需要手动释放；lock不是java内置特性。

package com.boot.demo;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author braska
 * @date 2020/3/9
 **/
public class LockTest {

    private Lock lock = new ReentrantLock();
    private AtomicInteger stock = new AtomicInteger(0);

    public void produce() {
        lock.lock();
        if (stock.get() < 5) {
            System.out.println(String.format("目前库存：%s", stock.addAndGet(1)));
        } else {
            System.out.println("仓库已满。");
        }
        lock.unlock();
    }

    public void consume(String name) {
        lock.lock();
        if (stock.get() > 0) {
            System.out.println(String.format("%s正在消费，目前库存: %s", name, stock.addAndGet(-1)));
        } else {
            System.out.println("库存已光。");
        }
        lock.unlock();
    }

    static class Producer implements Runnable{

        private LockTest deal;
        public Producer(LockTest deal) {
            this.deal = deal;
        }

        @Override
        public void run() {
            while (true)
                try {
                    Thread.sleep(1000);
                    deal.produce();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

        }
    }

    static class Consumer implements Runnable {
        private LockTest deal;
        private String name;
        public Consumer(LockTest deal, String name) {
            this.deal = deal;
            this.name = name;
        }

        @Override
        public void run() {
            while (true)
                try {
                    Thread.sleep(3000);
                    deal.consume(name);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
        }
    }


    public static void main(String[] args) {
        LockTest deal = new LockTest();

        Thread p = new Thread(new Producer(deal));

        Thread c1 = new Thread(new Consumer(deal, "消费者1"));
        Thread c2 = new Thread(new Consumer(deal, "消费者2"));

        p.start();
        c1.start();
        c2.start();
    }
}

四、BlockingQueue方式

底层采用的也是lock机制。

package com.boot.demo;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author braska
 * @date 2020/3/9
 **/
public class BlockingQueueTest {

    private BlockingQueue<Integer> queue = new ArrayBlockingQueue(5);
    private AtomicInteger stock = new AtomicInteger(0);

    public void produce() throws InterruptedException {
        try {
            queue.add(stock.addAndGet(1));
            System.out.println(String.format("目前库存：%s", stock.get()));
        } catch (Exception e) {
            stock.addAndGet(-1);
            System.out.println("仓库已满。");
        }
    }

    public void consume(String name) {
        Integer good = queue.poll();
        if (good != null) {
            System.out.println(String.format("%s正在消费，目前库存: %s", name, good));
            stock.addAndGet(-1);
        } else {
            System.out.println("库存已光。");
        }
    }

    static class Producer implements Runnable{

        private BlockingQueueTest deal;
        public Producer(BlockingQueueTest deal) {
            this.deal = deal;
        }

        @Override
        public void run() {
            while (true)
                try {
                    Thread.sleep(1000);
                    deal.produce();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

        }
    }

    static class Consumer implements Runnable {
        private BlockingQueueTest deal;
        private String name;
        public Consumer(BlockingQueueTest deal, String name) {
            this.deal = deal;
            this.name = name;
        }

        @Override
        public void run() {
            while (true)
                try {
                    Thread.sleep(3000);
                    deal.consume(name);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
        }
    }


    public static void main(String[] args) {
        BlockingQueueTest deal = new BlockingQueueTest();

        Thread p = new Thread(new Producer(deal));

        Thread c1 = new Thread(new Consumer(deal, "消费者1"));
        Thread c2 = new Thread(new Consumer(deal, "消费者2"));

        p.start();
        c1.start();
        c2.start();
    }
}