【Java 并发】【应用】经典的生产者、消费者

1  前言

闲来无事，复习复习并发中常用到的一些协调多线程的工具哈。

2  基于Java队列的实现

生产者跟消费者之间要协调，他俩会出现碰撞的地方就是存放东西的容器，所以我们可以直接拿一个线程安全的队列来做容器即可，比如我这里用的 ArrayBlockingQueue:

/**
 * @author: xjx
 * @description
 */
public class SyncDemo {

    /**
     * 容器大小
     */
    private int len = 10;
    /**
     * 队列
     */
    private ArrayBlockingQueue queue = new ArrayBlockingQueue(len);

    /**
     * 生产
     * add put(阻塞) offer(不阻塞）
     */
    public void product(int num) throws InterruptedException {
        queue.put(num);
        System.out.println(String.format("%s-生产了-%s", Thread.currentThread().getName(), num));
    }

    /**
     * 消费
     * poll(不阻塞) take(阻塞) peek(只看不取）
     */
    public int consume() throws InterruptedException {
        Object num = queue.take();
        System.out.println(String.format("%s-消费了-%s", Thread.currentThread().getName(), num));
        return ((int) num);
    }

    public static void main(String[] args) throws InterruptedException {
        SyncDemo demo = new SyncDemo();
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 100; i++) {
                    demo.consume();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Thread product = new Thread(() -> {
            try {
                for (int i = 0; i < 100; i++) {
                    demo.product(i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        product.start();
        consumer.start();

        product.join();
        consumer.join();
    }
}

3  基于Java内置锁的实现

那么我们不用安全队列的情况下，基于最原始的 Java 内置锁方式的实现如下：

/**
 * @author: xjx
 * @description
 */
public class SyncedDemo {

    /**
     * 容器大小
     */
    private int len = 10;
    /**
     * 容器
     */
    private List<Integer> list = Lists.newArrayListWithCapacity(len);
    /**
     * 同步对象
     */
    private final Object obj = new Object();

    /**
     * 生产
     * add put(阻塞) offer(不阻塞）
     */
    public void product(int num) throws InterruptedException {
        synchronized (obj) {
            while (list.size() == len) {
                obj.wait();
            }
            // 不满了 可以放东西了
            list.add(num);
            System.out.println(String.format("%s-生产了-%s", Thread.currentThread().getName(), num));
            // 唤醒消费者
            obj.notify();
        }
    }

    /**
     * 消费
     * poll(不阻塞) take(阻塞) peek(只看不取）
     */
    public void consume() throws InterruptedException {
        synchronized (obj) {
            while (list.size() <= 0) {
                obj.wait();
            }
            // 不空，消费一个
            Integer num = list.remove(list.size() - 1);
            System.out.println(String.format("%s-消费了-%s", Thread.currentThread().getName(), num));
            // 唤醒生产者
            obj.notify();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        SyncedDemo demo = new SyncedDemo();
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 100; i++) {
                    demo.consume();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Thread product = new Thread(() -> {
            try {
                for (int i = 0; i < 100; i++) {
                    demo.product(i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        product.start();
        consumer.start();

        product.join();
        consumer.join();
    }
}

4  基于AQS延申的同步工具类的实现

基于AQS 延申出很多同步工具类，我这里用 ReentrantLock 来实现下：

public class ReentrantSyncDemo {

    /**
     * 容器大小
     */
    private int len = 10;
    /**
     * 容器
     */
    private List<Integer> list = Lists.newArrayListWithCapacity(len);
    /**
     * 同步对象
     */
    private ReentrantLock lock = new ReentrantLock();
    private Condition notFull = lock.newCondition();
    private Condition notEmpty = lock.newCondition();

    /**
     * 生产
     * add put(阻塞) offer(不阻塞）
     */
    public void product(int num) throws InterruptedException {
        lock.lock();
        try {
            while (list.size() == len) {
                notFull.await();
            }
            // 不满了 可以放东西了
            list.add(num);
            System.out.println(String.format("%s-生产了-%s", Thread.currentThread().getName(), num));
            // 唤醒消费者
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 消费
     * poll(不阻塞) take(阻塞) peek(只看不取）
     */
    public void consume() throws InterruptedException {
        lock.lock();
        try {
            while (list.size() <= 0) {
                notEmpty.await();
            }
            // 不空，消费一个
            Integer num = list.remove(list.size() - 1);
            System.out.println(String.format("%s-消费了-%s", Thread.currentThread().getName(), num));
            // 唤醒生产者
            notFull.signal();
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ReentrantSyncDemo demo = new ReentrantSyncDemo();
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 100; i++) {
                    demo.consume();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Thread product = new Thread(() -> {
            try {
                for (int i = 0; i < 100; i++) {
                    demo.product(i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        product.start();
        consumer.start();

        product.join();
        consumer.join();
    }
}

5  小结

好啦，大概先想到这三种，有理解不对的地方欢迎指正哈。