Java面试——写一个生产者与消费者

Java架构师必看

发布于 2021-05-14 16:23:12

1.5K0

发布于 2021-05-14 16:23:12

一、通过synchronize 中的 wait 和 notify 实现

【1】我们可以将生产者和消费者需要的方法写在公共类中

package com.yintong.concurrent;

import java.util.LinkedList;

public class Concurrentcomm {
    //常量
    private static int MAX_VALUE = 10;
    //可以理解为缓存
    LinkedList<String> linkedList = new LinkedList<>();
    Object object = new Object();
    /*
     * 生产者方法
     */
    public void product() throws Exception {
        synchronized(linkedList) {
            while(MAX_VALUE == linkedList.size()) {
                System.out.println("仓库已满，【生产者】： 暂时不能执行生产任务!");
                linkedList.wait();
            }
            linkedList.push("   李四    ");
            System.out.println("【生产者】：生产了一个产品\t【现仓储量为】:" + linkedList.size());
            linkedList.notifyAll();
        }
    }
    /*
     * 消费者方法
     */
    public void customer() throws Exception {
        /*
         * 根据jdk的void notifyAll()的描述，“解除那些在该对象上调用wait()方法的线程的阻塞状态。该方法只能在同步方法或同步块内部调用。
         * 如果当前线程不是对象所得持有者，
         * 该方法抛出一个java.lang.IllegalMonitorStateException 异常”
         * so我们使用同一把锁
         */
        synchronized (linkedList) {
                        //多线程判断中使用 while 不要使用 if 否则会出现虚假唤醒问题
            while(linkedList.size() == 0) {
                System.out.println("仓库无货，【消费者】： 暂时不能执行消费任务!");
                linkedList.wait();
            }
            linkedList.pop();
            System.out.println("【消费者】：消费了一个产品\t【现仓储量为】:" + linkedList.size());
            linkedList.notifyAll();
        }
    }
}

【2】在 main 函数中调用生产者和消费者方法，并加限制即可

/**
 * @author zzx
 * @desc 生产者与消费者
 *
 */
public class Concurrent {
    //常量
    private static int MAX_VALUE = 100;

    public static void main(String[] args) {
        Concurrentcomm con = new Concurrentcomm();
        new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    for (int i = 0; i < MAX_VALUE; i++) {
                        Thread.sleep(0);
                        con.product();
                    }
                } catch (Exception e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }).start();
        // 消费者
        new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    Thread.sleep(10);
                    for (int i = 0; i < MAX_VALUE; i++) {
                        con.customer();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

【3】简单的生产者与消费者模式就完成了，可以看下运行的结果

二、通过 Lock 中的 await 与 signalAll 实现

【1】我们将公共的属性和方法放在 Resouce 类中，在资源类中使用 Lock 中的 lock()进行加锁，控制并发操作。使用 await()方法阻塞线程。使用 signalAll()唤醒线程。

/**
 * 通过 Lock 实现生产者与消费者
 * 资源类：将公共的资源放在一个单独的类中，可以将其看做一个产品，自身就就有生产和消费的能力（方法）
 */
public class ProductAndConsumer {
    public static void main(String[] args) {
        Resouce resouce = new Resouce();
        //生产者
        new Thread(()->{
            for (int i=1;i<=5;i++) {
                resouce.product();
            }
        },String.valueOf("生产者")) .start();

        //消费者
        new Thread(()->{
            for (int i=1;i<=5;i++){
                resouce.consumer();
            }
        },String.valueOf("消费者")).start();
    }
}
//资源类
class Resouce {
    private int MAX_VALUE = 3;
    private int MIN_VALUE = 0;
    private int number = 0;
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();

    //生产者
    public void product(){
        try {
            lock.lock();
            //如果生产的数量大于最大值则阻塞
            while(number >= MAX_VALUE){
                condition.await();
            }
            number++;
            System.out.println("【生产者】：生产了一个产品\t【现仓储量为】:" + number);
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }

    //消费者
    public void  consumer(){
        try {
            lock.lock();
            //如果消费的值=0则阻塞
            while(number <= MIN_VALUE){
                condition.await();
            }
            number--;
            System.out.println("【消费者】：消费了一个产品\t【现仓储量为】:" + number);
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
}

【2】输出结果展示：

三、synchronized 和 Lock 的区别

【1】原始构成：synchronized 是关键字属于 JVM 层面。底层通过 monitorenter（进入）monitorexit（退出）实现。底层是通过 monitor 对象完成，其实 wait/notify 等方法也依赖于 monitor 对象，只有在同步块或方法中才能调用 wait/notify 等方法。Lock 是具体类（java.util.concurrent.locks.Lock）是 API 层面的锁。【2】使用方法：synchronized 不需要用户手动释放锁，当 synchronized 代码执行完后，系统会自动释放锁。ReentrantLock 则需要用户手动释放锁，若未主动释放锁，就可能导致出现死锁的现象。【3】等待是否中断：synchronized 不可中断，除非抛出异常或者正常运行完成。ReentrantLock 可中断，1）、设置超时时间 tryLock(long timeout,TimeUnit unit) 2）、lockInterruptibly() 放在代码块中，调用 interrupt() 方法可中断。【4】加锁是否公平：synchronized 非公平锁。ReentrantLock 两者都可以，默认是非公平锁，构造方法可以传入 boolean 值，true 为公平锁，false 为非公平锁。【5】锁绑定多个条件 Condition：synchronized 没有。ReentrantLock 用来实现分组唤醒需要唤醒的线程们，可以精确唤醒，而不是像 synchronized 要么随机唤醒一个线程要么唤醒全部线程。

四、通过阻塞队列实现生产者与消费者

【1】通过blockQueue 中的 put/take 方法实现生产者与消费者，具体实现如下：当生产者使用put 生产到指定的队列大小3时，就会阻塞当前线程。这是消费者线程会通过 take 方法消费队列中的消息。当队列中没有消息时，会阻塞，直到有消息消费。

public class BlockProductConsumer {
    public static void main(String[] args) {
        MyResouce resouce = new MyResouce(new ArrayBlockingQueue(3));
        //生产者线程
        new Thread(()->{
            for(int i=1;i<=10;i++){
                resouce.product();
            }
        },"生产者").start();

        //消费者线程
        new Thread(()->{
            for(int i=1;i<=10;i++){
                try {
                    resouce.consumer();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"消费者").start();

        try {
            TimeUnit.SECONDS.sleep(1);
            resouce.stop();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}


/**
 * 公共资源类
 */
class MyResouce{
    //标记 while 无限循环
    private volatile boolean FLAG = true;
    //队列中存入的数值
    private AtomicInteger atomicInteger = new AtomicInteger();
    //组合一个阻塞队列，通过构造器传入
    private BlockingQueue blockingQueue;
    public MyResouce(BlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    //生产者
    public void product(){
        try {
            while (FLAG){
                blockingQueue.put(String.valueOf(atomicInteger.incrementAndGet()));
                System.out.println("生产者生产第"+blockingQueue.size()+"个产品");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //消费者
    public void consumer() throws InterruptedException {
        while (FLAG){
            blockingQueue.take();
            System.out.println("消费者消费第"+(blockingQueue.size()+1)+"个产品");
        }
    }

    public void stop(){
        FLAG = false;
        System.out.println("========================");
    }
}

【2】效果展示：