线程池(阻塞队列,生产者消费者模式)

线程池(阻塞队列,生产者消费者模式),第1张

线程池(阻塞队列,生产者消费者模式) 线程等待和通知

Object类中有些方法不可随便调用
wait()、notify()、notifyAll()这三个方法:
1.wait() 使线程主动释放锁,进入等待状态,直到他被其他的线程通过 notify() 和 notifyAll() 唤醒 或者 超过等待时间
2.wait(long) 让当前线程进入等待状态,同时设置时间;直到被通知为止或时间结束
3.notify() 随机通知一个等待线程
4.notifyAll() 通知所有的等待线程

注意:等待和通知方法必须是锁对象,否则会抛出IllegalMonitorStateException

public class WaitDemo {

    public synchronized void waiteTest() throws InterruptedException {
        //当前方法等待
        System.out.println(Thread.currentThread().getName() + "当前方法等待5s");
        this.wait();
    }

    public synchronized void notifyTest(){
        System.out.println("唤醒等待中的方法");
        this.notify();
    }

    public static void main(String[] args) {
        WaitDemo waitDemo = new WaitDemo();
        Thread thread = new Thread(() -> {
            try {
                waitDemo.waiteTest();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        thread.start();
        try {
            Thread.sleep(5000);
            waitDemo.notifyTest();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
wait()和sleep()区别
  1. 调用对象不同

    wait() 由锁对象调用

    sleep() 由线程调用

  2. 锁使用不同

    执行wait后,自动释放锁

    执行sleep后,不会释放锁

  3. 唤醒机制不同

    执行wait后,可以被通知唤醒

    执行sleep后,只能等待时间结束后,自动唤醒

生产者消费者模式

例子:假如一个包子铺卖包子,应该是一边生产,一边销售,工作效率最高。如果来了一群顾客,这时候开始一个个生产效率太低,所以顾客还没来就要开始生产,这就需要一个蒸笼,把生产好的包子放在蒸笼里,顾客来了就卖给顾客。

生产者:生产数据的线程或者进程
消费者:使用数据的线程或进程

而蒸笼就相当于一个缓存区
模式实现思路:
1.添加缓冲区,设置上限
2.向缓存区存放数据,满了使生产者进入等待,有空位置则生产
3.消费者从缓冲区使用数据,缓存区没有数据则消费者等待,有数据再通知消费者

解决问题

  • 为生产和消费解耦
  • 提高并发性能
  • 解决忙闲不均
public class BaoziDemo0 {
    public static void main(String[] args) {
        BaoziShop baoziShop = new BaoziShop();
        //创建一个线程,用来生产
        for (int i = 0; i < 150; i++) {
            new Thread(()->{
                try {
                    baoziShop.produceBao(baoziShop.baozis.size() + 1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }

        //创建一个线程,用来消费
        for (int i = 0; i < 10; i++) {
            new Thread(()->{
                try {
                    for (int j = 0; j < 20; j++) {
                        baoziShop.sellBao();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

//包子铺
class BaoziShop{

    //缓冲区 存放数据
    List baozis = new ArrayList<>();

    //上限
    int MAX_COUNT = 100;

    //生产包子
    public synchronized void produceBao(int id) throws InterruptedException {
        //如果缓存区满了
        if (baozis.size() == MAX_COUNT){
            System.out.println("缓存区满了---------------------" + Thread.currentThread().getName()+"等待");
            this.wait();
        }else {
            //通知卖包子线程
            this.notify();
        }
        BaoZi baoZi = new BaoZi(baozis.size() + 1);
        baozis.add(baoZi);
        System.out.println( Thread.currentThread().getName() + "师傅生产一个包子" + baoZi);
    }

    //卖包子
    public synchronized void sellBao() throws InterruptedException {
        //如果缓存区没包子了
        if (baozis.size() == 0){
            System.out.println("缓存区空了--------------" +Thread.currentThread().getName()+"等待");
            this.wait();
        }else {
        //否则,通知生产包子
            this.notify();
        }
        if (baozis.size()>0){
            BaoZi baoZi = baozis.remove(0);
            System.out.println("顾客买了一个包子" + baoZi);
        }
    }
}

//包子类
class BaoZi{
    public Integer id;

    public BaoZi(Integer id) {
        this.id = id;
    }

    @Override
    public String toString() {
        return "BaoZi-"+ id;
    }
}
阻塞队列

应用了生产者消费者模式的集合,能够根据数据满或空的情况,自动对线程执行等待和通知

BlockingQueue接口

  • put 添加数据,达到线程上限会让线程自动等待
  • take 取并删除数据,数据空了会让线程自动等待
    实现类
    ArrayBlockingQueue类 数组
    linkedBlockingQueue类 链表结构
public class BaoziDemo {
    
    static class Baozi{
        private int id;
        public Baozi(int id) {
            this.id = id;
        }

        @Override
        public String toString() {
            return "包子--" + id;
        }
    }

    public static void main(String[] args) {
        //阻塞队列
        BlockingQueue baozis = new ArrayBlockingQueue<>(100);
        //生产者线程
        new Thread(() -> {
            for (int i = 0; i < 200; i++) {
                //创建包子,添加到阻塞队列,满了就自动阻塞线程
                Baozi baozi = new Baozi(baozis.size() + 1);
                try {
                    baozis.put(baozi);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName()+"生产了" + baozi);
            }
        }).start();
        //消费者线程
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                for (int j = 0; j < 40; j++) {
                    //取包子取完了会自动阻塞
                    try {
                        Baozi take = baozis.take();
                        System.out.println(Thread.currentThread().getName() + "消费了" + take);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}
线程池

我们创建线程在运行结束后都会被虚拟机销毁,如果线程数量多的话,频繁的创建和销毁线程会大大浪费时间和效率,更重要的是浪费内存。那么有没有一种方法能让线程运行完后不立即销毁,而是让线程重复使用,继续执行其他的任务?

顶层接口:Executor

  • execute(Runnable) 启动线程执行一个任务

ExecutorService 继承 Executor 添加了线程池管理方法,如:shutdown()、shutdownNow()

ThreadPoolExecuter作为线程池的实现类 继承了实现 ExecutorService接口的抽象类AbstractExecutorService

  • corePoolSize 核心线程数,创建线程池后自带线程,不会进行销毁
  • maximumPoolSize 最大线程数
  • keepAliveTime 存活时间,非核心线程能够闲置的时间,超过后被销毁
  • timeUnit 时间单位
  • blockingQueue 阻塞队列 存放任务(Runnable)的集合

优化配置

  1. 核心线程数 应该和CPU内核数量相关 CPU内核数 * N (N和任务执行需要时间和并发量相关)
  2. 最大线程数可以和核心线程数一样,避免频繁创建和销毁线程
  3. 如果存在非核心线程,设置大一点,避免频繁创建和销毁线程
  4. 阻塞队列使用linkedBlockingQueue,插入和删除任务效率更高
public class ThreadPoolExecutorDemo {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService threadPoolExecutor =
                new ThreadPoolExecutor(2, 5,
                        5000L, TimeUnit.SECONDS, new linkedBlockingDeque() {
                });
        for (int i = 0;i<50; i++) {
            int n = i+1;
            threadPoolExecutor.execute(()->{
                System.out.println(Thread.currentThread().getName() + "执行了任务" + n);
            });
        }
    }
}

线程池原理
预先启动一些线程,线程无限循环从任务队列中获取一个任务进行执行,直到线程池被关闭。如果某个线程因为执行某个任务发生异常而终止,那么重新创建一个新的线程而已,如此反复。

工作线程的集合都在一个存储work对象的HashSet中,线程池的工作线程通过Woker类实现,通过ReentrantLock锁保证线程安全。添加线程到workers中(线程池中)。 多余的任务会放在阻塞队列中。只有当阻塞队列满了后,才会触发非核心线程的创建。所以非核心线程只是临时过来打杂的。直到空闲了,然后自己关闭了。

欢迎分享,转载请注明来源:内存溢出

原文地址: https://outofmemory.cn/zaji/5660773.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2022-12-16
下一篇 2022-12-16

发表评论

登录后才能评论

评论列表(0条)

保存