java生产者消费者模型

2023-05-16

前言

生产者和消费者问题是线程模型中的经典问题：生产者和消费者在同一时间段内共用同一个存储空间，生产者往存储空间中添加产品，消费者从存储空间中取走产品，当存储空间为空时，消费者阻塞，当存储空间满时，生产者阻塞。

基于synchronized配合wait()和notify()方法的实现

这也是最简单最基础的实现，缓冲区满和为空时都调用wait()方法等待，当生产者生产了一个产品或者消费者消费了一个产品之后会唤醒所有线程。


public class Test1 {
    private static Integer count = 0;
    private static final Integer FULL = 10;
    private static String LOCK = "lock";
    
    public static void main(String[] args) {
        Test1 test1 = new Test1();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                synchronized (LOCK) {
                    while (count == FULL) {
                        try {
                            LOCK.wait();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    count++;
                    System.out.println(Thread.currentThread().getName() + "生产者生产，目前总共有" + count);
                    LOCK.notifyAll();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (LOCK) {
                    while (count == 0) {
                        try {
                            LOCK.wait();
                        } catch (Exception e) {
                        }
                    }
                    count--;
                    System.out.println(Thread.currentThread().getName() + "消费者消费，目前总共有" + count);
                    LOCK.notifyAll();
                }
            }
        }
    }
}

结果：

Thread-0生产者生产，目前总共有1
Thread-4生产者生产，目前总共有2
Thread-3消费者消费，目前总共有1
Thread-1消费者消费，目前总共有0
Thread-2生产者生产，目前总共有1
Thread-6生产者生产，目前总共有2
Thread-7消费者消费，目前总共有1
Thread-5消费者消费，目前总共有0
Thread-0生产者生产，目前总共有1
Thread-4生产者生产，目前总共有2
Thread-3消费者消费，目前总共有1
Thread-6生产者生产，目前总共有2
Thread-1消费者消费，目前总共有1
Thread-7消费者消费，目前总共有0
Thread-2生产者生产，目前总共有1
Thread-5消费者消费，目前总共有0
Thread-0生产者生产，目前总共有1
Thread-4生产者生产，目前总共有2
Thread-3消费者消费，目前总共有1
Thread-7消费者消费，目前总共有0
Thread-6生产者生产，目前总共有1
Thread-2生产者生产，目前总共有2
Thread-1消费者消费，目前总共有1
Thread-5消费者消费，目前总共有0
Thread-0生产者生产，目前总共有1
Thread-4生产者生产，目前总共有2
Thread-3消费者消费，目前总共有1
Thread-1消费者消费，目前总共有0
Thread-6生产者生产，目前总共有1
Thread-7消费者消费，目前总共有0
Thread-2生产者生产，目前总共有1

基于可重入锁ReentrantLock结合Condition的实现

public class Test2 {
    private static Integer count = 0;
    private static final Integer FULL = 10;
    //创建一个锁对象
    private Lock lock = new ReentrantLock();
    //创建两个条件变量，一个为缓冲区非满，一个为缓冲区非空
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    public static void main(String[] args) {
        Test2 test2 = new Test2();
        new Thread(test2.new Producer()).start();
        new Thread(test2.new Consumer()).start();
        new Thread(test2.new Producer()).start();
        new Thread(test2.new Consumer()).start();
        new Thread(test2.new Producer()).start();
        new Thread(test2.new Consumer()).start();
        new Thread(test2.new Producer()).start();
        new Thread(test2.new Consumer()).start();
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                //获取锁
                lock.lock();
                try {
                    while (count == FULL) {
                        try {
                            notFull.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    count++;
                    System.out.println(Thread.currentThread().getName()
                            + "生产者生产，目前总共有" + count);
                    //唤醒消费者
                    notEmpty.signal();
                } finally {
                    //释放锁
                    lock.unlock();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                lock.lock();
                try {
                    while (count == 0) {
                        try {
                            notEmpty.await();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    count--;
                    System.out.println(Thread.currentThread().getName()
                            + "消费者消费，目前总共有" + count);
                    notFull.signal();
                } finally {
                    lock.unlock();
                }
            }
        }
    }
}

基于阻塞队列BlockingQueue的实现

BlockingQueue即阻塞队列，当一个线程对已经满了的阻塞队列进行入队操作时会阻塞，除非有另外一个线程进行了出队操作，当一个线程对一个空的阻塞队列进行出队操作时也会阻塞，除非有另外一个线程进行了入队操作。

阻塞队列是线程安全的。

BlockingQueue接口的一些方法：

操作	抛异常	特定值	阻塞	超时
插入	add(o)	offer(o)	put(o)	offer(o, timeout, timeunit)
移除	remove(o)	poll(o)	take(o)	poll(timeout, timeunit)
检查	element(o)	peek(o)

这四类方法分别对应的是：

        1 . ThrowsException：如果操作不能马上进行，则抛出异常。
        2 . SpecialValue：如果操作不能马上进行，将会返回一个特殊的值，一般是true或者false。
        3 . Blocks:如果操作不能马上进行，操作会被阻塞。
        4 . TimesOut:如果操作不能马上进行，操作会被阻塞指定的时间，如果指定时间没执行，则返回一个特殊值，一般是true或者false。

public class ConsumerAndProducer {
    //创建一个阻塞队列
    final BlockingQueue blockingQueue = new ArrayBlockingQueue<>(10);
    final Random random = new Random();
    public static void main(String[] args) {
        ConsumerAndProducer test3 = new ConsumerAndProducer();
        new Thread(test3.new Producer()).start();
        new Thread(test3.new Consumer()).start();
        new Thread(test3.new Producer()).start();
        new Thread(test3.new Consumer()).start();
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 5; i++) {
                try {
                    Thread.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                try {
                    Double d = random.nextDouble();
                    blockingQueue.put(d);
                    System.out.println(Thread.currentThread().getName()
                            + "生产者生产:" + d);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 5; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                try {
                    Object take = blockingQueue.take();
                    System.out.println(Thread.currentThread().getName()
                            + "消费者消费:" + take);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}

结果；

Thread-1消费者消费:0.32443113589507067
Thread-3消费者消费:0.9632195811796581
Thread-0生产者生产:0.9632195811796581
Thread-2生产者生产:0.32443113589507067
Thread-3消费者消费:0.6215293059181483
Thread-0生产者生产:0.6215293059181483
Thread-2生产者生产:0.3358615332068282
Thread-1消费者消费:0.3358615332068282
Thread-3消费者消费:0.29797328615872753
Thread-0生产者生产:0.29797328615872753
Thread-2生产者生产:0.8414919052915146
Thread-1消费者消费:0.8414919052915146
Thread-3消费者消费:0.4963327348671358
Thread-0生产者生产:0.4963327348671358
Thread-2生产者生产:0.43976197901390424
Thread-1消费者消费:0.43976197901390424
Thread-3消费者消费:0.32807845639243016
Thread-0生产者生产:0.32807845639243016
Thread-2生产者生产:0.01487664860711524
Thread-1消费者消费:0.01487664860711524

基于信号量Semaphore的实现

Semaphore（信号量）是用来控制同时访问特定资源的线程数量，通过协调各个线程以保证合理地使用公共资源。

Semaphore通过使用计数器来控制对共享资源的访问。如果计数器大于0，则允许访问。如果为0，则拒绝访问。计数器所计数的是允许访问共享资源的许可。因此，要访问资源，必须从信号量中授予线程许可。

void acquire() ：从信号量获取一个许可，如果无可用许可前将一直阻塞等待。

void acquire(int permits) ：获取指定数目的许可，如果无可用许可前也将会一直阻塞等待

boolean tryAcquire()：从信号量尝试获取一个许可，如果无可用许可，直接返回false，不会阻塞

boolean tryAcquire(int permits)：尝试获取指定数目的许可，如果无可用许可直接返回false

boolean tryAcquire(int permits, long timeout, TimeUnit unit)：
在指定的时间内尝试从信号量中获取许可，如果在指定的时间内获取成功，返回true，否则返回false

void release()：释放一个许可，别忘了在finally中使用，注意：多次调用该方法，会使信号量的许可数增加，达到动态扩展的效果，如：初始permits为1，调用了两次release，最大许可会改变为2

int availablePermits()：获取当前信号量可用的许可

Java中的Semaphore维护了一个许可集，一开始先设定这个许可集的数量，可以使用acquire()方法获得一个许可，当许可不足时会被阻塞，release()添加一个许可。在下列代码中，还加入了另外一个mutex信号量，维护生产者消费者之间的同步关系，保证生产者和消费者之间的交替进行。

public class ConsumerAndProducer {

    private static Integer count = 0;
    //创建三个信号量
    final Semaphore notFull = new Semaphore(10);
    //notEmpty的入参为0表示，第一次进行acquire就会被阻塞，直到进行了release操作，释放了一个许可
    final Semaphore notEmpty = new Semaphore(0);
    final Semaphore mutex = new Semaphore(1);
    public static void main(String[] args) {
        ConsumerAndProducer test4 = new ConsumerAndProducer();
        new Thread(test4.new Producer()).start();
        new Thread(test4.new Consumer()).start();
        new Thread(test4.new Producer()).start();
        new Thread(test4.new Consumer()).start();
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 5; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    notFull.acquire();
                    mutex.acquire();
                    count++;
                    System.out.println(Thread.currentThread().getName()
                            + "生产者生产，目前总共有" + count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    mutex.release();
                    notEmpty.release();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 5; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                try {
                    notEmpty.acquire();
                    mutex.acquire();
                    count--;
                    System.out.println(Thread.currentThread().getName()
                            + "消费者消费，目前总共有" + count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    mutex.release();
                    notFull.release();
                }
            }
        }
    }
}

运行结果：

Thread-2生产者生产，目前总共有1
Thread-0生产者生产，目前总共有2
Thread-1消费者消费，目前总共有1
Thread-3消费者消费，目前总共有0
Thread-0生产者生产，目前总共有1
Thread-2生产者生产，目前总共有2
Thread-1消费者消费，目前总共有1
Thread-3消费者消费，目前总共有0
Thread-0生产者生产，目前总共有1
Thread-1消费者消费，目前总共有0
Thread-2生产者生产，目前总共有1
Thread-3消费者消费，目前总共有0
Thread-2生产者生产，目前总共有1
Thread-3消费者消费，目前总共有0
Thread-0生产者生产，目前总共有1
Thread-1消费者消费，目前总共有0
Thread-2生产者生产，目前总共有1
Thread-0生产者生产，目前总共有2
Thread-3消费者消费，目前总共有1
Thread-1消费者消费，目前总共有0

参考文档：

Java生产者消费者模型的五种实现方式_wowwilliam0的博客-CSDN博客_java生产者消费者模型

Java Semaphore详解_warybee-CSDN博客_java semaphore

生产者消费者模式-Java实现 - 天目山电鳗 - 博客园

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)