Java--多线程之生产者消费者模式；线程池ExecutorService

Posted 2022-10-15 MinggeQingchun

Java--多线程之并发，并行，进程，线程（一）_MinggeQingchun的博客-CSDN博客

Java--多线程之终止/中断线程（二）_MinggeQingchun的博客-CSDN博客_java中断线程 

Java--多线程之join，yield，sleep；线程优先级；定时器；守护线程（三）_MinggeQingchun的博客-CSDN博客 

Java--多线程之synchronized和lock；死锁（四）_MinggeQingchun的博客-CSDN博客 

一、生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
这个阻塞队列就是用来给生产者和消费者解耦的。纵观大多数设计模式，都会找一个第三者出来进行解耦，如工厂模式的第三者是工厂类，模板模式的第三者是模板类

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这种生产消费能力不均衡的问题，所以便有了生产者和消费者模式

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度

Java中生产者消费者模式主要使用wait()方法和notify()方法

方法名	作用
wait()	表示线程一直等待，直到其他线程通知，与sleep不同，会释放锁
wait(long timeout)	指定等待的毫秒数
notify()	唤醒一个处于等待状态的线程；不会释放占有的锁
notifyAl()	唤醒同一个对象上所有调用wait()方法的线程，优先级比较高的线程优先调度

1、管程法

/**
 1、“生产者和消费者模式“
    生产线程负责生产，消费线程负责消费
    生产线程和消费线程要达到均衡
    这种特殊的业务需求需要使用wait()方法和notify()方法
 2、wait()和notify()方法不是线程对象的方法，是普通java对象都有的方法
 3、wait()方法和notify()方法建立在线程同步的基础之上；多线程要同时操作一个仓库。有线程安全问题
 4、wait()方法作用：obj.wait()让正在obj对象上活动的线程t进入等待状态，并且释放掉t线程之前占有的obj对象的锁
 5、notify()方法作用：obj.notify()让正在obj对象上等待的线程唤醒，只是通知，不会释放obj对象上之前占有的锁
 */
public class ProducerConsumerModel 
    public static void main(String[] args) 
        /**需求：
        List集合中假设只能存储1个元素
        如果List集合中元素个数是0，就表示仓库空了
        保证List集合中永远都是最多存储1个元素
        要求：生产1个消费1个
         */
        List list = new ArrayList();
        Thread t1 = new Thread(new ProducerThread(list));
        Thread t2 = new Thread(new ConsumerThread(list));

        t1.setName("生产者线程");
        t2.setName("消费者线程");

        t1.start();
        t2.start();
    


//生产者
class ProducerThread implements Runnable 
    private List list;

    public ProducerThread(List list)
        this.list = list;
    

    @Override
    public void run() 
        while (true)
            //使用死循环来模拟一直生产
            synchronized (list)
                if (list.size() > 0)// 大于0，说明list中已经有1个元素，不再生产
                    try 
                        // 当前线程进入等待状态，并且释放Producer之前占有的list集合
                        list.wait();
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                
                // list是空的，可以生产
                Object obj = new Object();
                list.add(obj);
                System.out.println(Thread.currentThread().getName() + "--->" + obj);
                // 唤醒消费者进行消费
                list.notify();
            
        
    


//消费者
class ConsumerThread implements Runnable 
    private List list;

    public ConsumerThread(List list)
        this.list = list;
    

    @Override
    public void run() 
        while (true)
            synchronized (list)
                if (list.size() == 0)// 等于0，说明list中没有元素，不再消费
                    try 
                        // 消费者线程等待，释放掉list集合的锁
                        list.wait();
                     catch (InterruptedException e) 
                        e.printStackTrace();
                    
                
                // list中有数据，进行消费
                Object obj = list.remove(0);
                System.out.println(Thread.currentThread().getName() + "--->" + obj);
                // 唤醒生产者生产
                list.notify();
            
        
    

2、信号灯法 

/**
 1、“生产者和消费者模式“
    生产线程负责生产，消费线程负责消费
    生产线程和消费线程要达到均衡
    这种特殊的业务需求需要使用wait()方法和notify()方法
 2、wait()和notify()方法不是线程对象的方法，是普通java对象都有的方法
 3、wait()方法和notify()方法建立在线程同步的基础之上；多线程要同时操作一个仓库。有线程安全问题
 4、wait()方法作用：obj.wait()让正在obj对象上活动的线程t进入等待状态，并且释放掉t线程之前占有的obj对象的锁
 5、notify()方法作用：obj.notify()让正在obj对象上等待的线程唤醒，只是通知，不会释放obj对象上之前占有的锁
 */
public class ProducerConsumerSignal 
    public static void main(String[] args) 
        /**需求：
        List集合中假设只能存储1个元素
        如果List集合中元素个数是0，就表示仓库空了
        保证List集合中永远都是最多存储1个元素
        要求：生产1个消费1个
         */
        List list = new ArrayList();
        Thread t1 = new Thread(new ProducerThread(list));
        Thread t2 = new Thread(new ConsumerThread(list));

        t1.setName("生产者线程");
        t2.setName("消费者线程");

        t1.start();
        t2.start();
    


//生产者
class ProducerThread1 implements Runnable 
    private Ware ware;

    public void ConsumerThread1(Ware ware)
        this.ware = ware;
    

    @Override
    public void run() 
        while (true)
            this.ware.product();
        
    


//消费者
class ConsumerThread1 implements Runnable 
    private Ware ware;

    public ConsumerThread1(Ware ware)
        this.ware = ware;
    

    @Override
    public void run() 
        while (true)
            this.ware.consumer();
        
    


class Ware 
    private List list;
    boolean flag = true;

    //生产
    public synchronized void product()
        if (!flag)// 大于0，说明list中已经有1个元素，不再生产
            try 
                // 当前线程进入等待状态，并且释放Producer之前占有的list集合
                list.wait();
             catch (InterruptedException e) 
                e.printStackTrace();
            
        
        // list是空的，可以生产
        Object obj = new Object();
        list.add(obj);
        System.out.println(Thread.currentThread().getName() + "--->" + obj);
        // 唤醒消费者进行消费
        list.notify();
    

    //消费
    public synchronized void consumer()
        if (flag)// 等于0，说明list中没有元素，不再消费
            try 
                // 消费者线程等待，释放掉list集合的锁
                list.wait();
             catch (InterruptedException e) 
                e.printStackTrace();
            
        
        // list中有数据，进行消费
        Object obj = list.remove(0);
        System.out.println(Thread.currentThread().getName() + "--->" + obj);
        // 唤醒生产者生产
        list.notify();
    

sleep和wait区别

1、sleep是Thread类的静态方法；wait是Object的方法

sleep的作用是让线程休眠指定的时间，在时间到达时恢复，也就是说sleep将在接到时间到达时恢复线程执行

wait是可以对任意一个对象调用wait方法，调用wait方法将会将调用者的线程挂起，直到其他线程调用同一个对象的notify方法才会重新激活调用者

2、sleep方法没有释放锁，而wait方法释放了锁

3、wait，notify和notifyAll只能在同步代码块里使用，而sleep可以在任何地方使用

synchronized(x)
      x.wait();
      //或者x.notify();

4、sleep和wait都需要捕获InterruptedException异常

二、线程池ExecutorService

JDK 5.0起提供了线程池相关API: ExecutorService 和 Executors

1、ExecutorService

真正的线程池接口。常见子类ThreadPoolExecutor

// 执行任务/命令,没有返回值，-般用来执行Runnable
void execute(Runnable command)

// 执行任务，有返回值，一般又来执行Callable
<T> Future<T> submit(Callable<T> task)

// 关闭连接池
void shutdown()

2、Executors

工具类、线程池的工厂类,用于创建并返回不同类型的线程池

3、ThreadPoolExecutor一共有七个参数

1、corePoolSize：核心线程数量

2、maximumPoolSize：线程最大数

3、workQueue：阻塞队列，存储等待执行的任务，会对线程池运行过程产生重大影响

当我们提交一个新的任务到线程池，线程池会根据当前池中正在运行的线程数量来决定该任务的处理方式

处理方式有三种：

（1）直接切换（SynchronusQueue）

（2）无界队列（LinkedBlockingQueue）能够创建的最大线程数为corePoolSize,这时maximumPoolSize就不会起作用了。当线程池中所有的核心线程都是运行状态的时候，新的任务提交就会放入等待队列中

（3）有界队列（ArrayBlockingQueue）最大maximumPoolSize，能够降低资源消耗，但是这种方式使得线程池对线程调度变的更困难。因为线程池与队列容量都是有限的。所以想让线程池的吞吐率和处理任务达到一个合理的范围，又想使我们的线程调度相对简单，并且还尽可能降低资源的消耗，我们就需要合理的限制这两个数量

分配技巧： 如果想降低资源的消耗包括降低cpu使用率、操作系统资源的消耗、上下文切换的开销等等，可以设置一个较大的队列容量和较小的线程池容量，这样会降低线程池的吞吐量。如果我们提交的任务经常发生阻塞，我们可以调整maximumPoolSize。如果我们的队列容量较小，我们需要把线程池大小设置的大一些，这样cpu的使用率相对来说会高一些。但是如果线程池的容量设置的过大，提高任务的数量过多的时候，并发量会增加，那么线程之间的调度就是一个需要考虑的问题。这样反而可能会降低处理任务的吞吐量

4、keepAliveTime：线程没有任务执行时最多保持多久时间终止（当线程中的线程数量大于corePoolSize的时候，如果这时没有新的任务提交核心线程外的线程不会立即销毁，而是等待，直到超过keepAliveTime）

5、unit：keepAliveTime的时间单位

6、threadFactory：线程工厂，用来创建线程，有一个默认的工场来创建线程，这样新创建出来的线程有相同的优先级，是非守护线程、设置好了名称）

7、rejectHandler：当拒绝处理任务时(阻塞队列满)的策略（AbortPolicy默认策略直接抛出异常、CallerRunsPolicy用调用者所在的线程执行任务、DiscardOldestPolicy丢弃队列中最靠前的任务并执行当前任务、DiscardPolicy直接丢弃当前任务）

corePoolSize、maximumPoolSize、workQueue 三者关系：

1、如果运行的线程数小于corePoolSize的时候，直接创建新线程来处理任务。即使线程池中的其他线程是空闲的

2、如果运行中的线程数大于corePoolSize且小于maximumPoolSize时，那么只有当workQueue满的时候才创建新的线程去处理任务

3、如果corePoolSize与maximumPoolSize是相同的，那么创建的线程池大小是固定的。这时有新任务提交，当workQueue未满时，就把请求放入workQueue中。等待空线程从workQueue取出任务。如果workQueue此时也满了，那么就使用另外的拒绝策略参数去执行拒绝策略

其他方法

execute();	//提交任务，交给线程池执行	
submit();//提交任务，能够返回执行结果 execute+Future
shutdown();//关闭线程池，等待任务都执行完
shutdownNow();//关闭线程池，不等待任务执行完
getTaskCount();//线程池已执行和未执行的任务总数
getCompleteTaskCount();//已完成的任务数量
getPoolSize();//线程池当前的线程数量
getActiveCount();//当前线程池中正在执行任务的线程数量

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class PoolExecutorTest 
    public static void main(String[] args) 
        // 1、创建线程池
        // newFixedThreadPool 参数为线程池大小
        ExecutorService service = Executors.newFixedThreadPool(10);

        // 执行
        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());

        // 2、关闭链接
        service.shutdown();
    


class MyThread implements Runnable
    @Override
    public void run() 
        System.out.println(Thread.currentThread().getName());