前言本文面向编程零基础小白用生活化案例通俗讲解 Java 中阻塞队列、线程池核心概念、组成要素与完整实操流程手把手演示阻塞队列与线程池的完整可运行代码示例。一、核心概念线程线程是程序里的一条“执行流”。 一个进程中可以有多条线程它们共享进程的内存 但各自可以有独立的栈和程序计数器。队列很多数据结构都可以属于队列。队列可以用部分数据结构实现比如数组、链表就是很经典的队列实现。队列需要满足先进先出、后进后出的条件。注意如果后进先出这是栈的特性并非队列。阻塞马路上的红绿灯亮起红灯时车必须停下等待这就是阻塞。在程序里当条件不满足时比如队列空了线程想取数据或者条件太满时比如队列满了线程想放数据就强制让当前线程停下来等一等。线程在等待期间不占用 CPU直到其他线程让条件满足了比如消费者取走了数据或者生产者放入了数据再把它叫醒继续干活。二、什么是线程池线程池就是“一组复用的线程”传统的线程往往只被赋予一段固定的功能新建出来开始工作工作结束之后直接关闭。但新建线程与关闭线程等操作对内存的消耗巨大一两次也许没事一旦数量上去了对内存的开销十分巨大。如果线程能在一项工作结束后并不直接关闭而是进入等待状态有了新的工作就不必再新建一个线程可以直接唤醒正在等待的线程让它继续新的工作。这样就可以避免频繁新建与关闭线程巨大的内存开销。想要实现这样的效果具体的“工作内容”并不能写在线程中线程只负责执行具体执行什么由外界传入。这种不会轻易关闭的线程一般被称为核心线程。当同一时间有多个任务需要执行时虽然可以等核心线程处理完前一个工作后再处理另一个工作但也可以再新建一个线程让新建的线程去处理工作提升整体效率。但是当线程的数量多了起来对内存的开销也会增加所以我们可以在需要时创建使用超时方法的临时线程为我们工作。当工作多到几个核心线程处理不过来时新建几个临时线程负担工作临时线程与核心线程一样并不在处理完一项工作后关闭而是一项接一项地处理各种不同的工作。但不同的是在没有工作进入等待状态时临时线程调用的超时方法会进行计时如果等待一定时间还没有被唤醒工作的话才会关闭以节省内存。由核心线程、临时线程、管理逻辑等等组成的一套系统就被称为线程池。三、什么是阻塞队列阻塞队列就是“有锁与阻塞机制的缓冲区或队列”在我写过的笔记“生产者-消费者模型” 与 “线程池” 当中我们已经见过了阻塞队列Buffer 类、 TaskList 类为了演示生产消费以及线程池复用线程的功能我用由synchronized修饰过的方法制作了一个简单的阻塞队列。阻塞队列使生产者往库里生产数据时不至于在库已满时依旧不停生产或使消费者不至于在库已空时继续不停消费。这次我学习 Java 官方的阻塞队列制作了一个基于锁ReentrantLock与缓冲区Condition的有基本功能的阻塞队列。单独使用锁ReentrantLock和缓冲区Condition与只使用synchronized和wait、notifyAll的区别在于前者可以存在多个缓冲区供不同职责的线程等待对应缓冲区调用await方法且依照先进先出的顺序唤醒其中的线程对应缓冲区调用signal方法。而wait只能使当前线程进入等待且与其他等待中线程在同一个缓冲区中无法区分线程的职责因此必须用notifyAll唤醒所有线程对应职责的那个线程可以进入工作其他的线程就白被唤醒一次只能回去重睡。因此使用缓冲区Condition进行分类和精准唤醒可以避免不必要的操作在一定程度上提升了效率。并且在某些情况下我们并不希望所有线程都被唤醒或是需要有顺序的唤醒线程就可以使用由“缓冲区Condition”实现的队列借助队列先进先出的特性达到要求。我创建了锁ReentrantLock的对象以便通过对象调用它的方法并设置了两个缓冲区一个是当任务栏满时让生产任务的线程等待的缓冲区另一个是当任务栏空时让处理任务的线程进入等待的缓冲区。注释总中我使用了 生产-消费 的命名生产与消费工作任务privateReentrantLocklocknewReentrantLock();//创建锁对象privateConditionfulllock.newCondition();//生产者等待区队列满时privateConditionemptylock.newCondition();//消费者等待区队列空时锁ReentrantLock的作用与synchronized类似可以通过lock方法避免在同一瞬间多个线程调用方法导致的数据紊乱。但这次代码选择使用锁ReentrantLock是为了借助ReentrantLock类中的awaitNanos方法来实现超时判断。这个方法可以让线程在等待时计算等待时间时间到时就可以直接return null;返回结束并关闭临时线程。//带超时的阻塞获取publicTpoll(longtimeOut,TimeUnitunit){//传进等待时间和时间单位lock.lock();try{longnanosunit.toNanos(timeOut);//统一转成纳秒while(queue.isEmpty()){System.out.println(Thread.currentThread().getName():队列已空进入等待…);if(nanos0){//等待时间已到returnnull;//直接返回}//等待指定时间//中途被唤醒会返回剩余时间//下次从记录时间继续等待nanosempty.awaitNanos(nanos);}Ttaskqueue.removeFirst();full.signal();returntask;}catch(InterruptedExceptione){thrownewRuntimeException(e);}finally{lock.unlock();}}}于是在需要暂时增加线程的时候创建的临时线程就可以通过调用这个方法在任务繁多时缓解压力在没有任务时光荣下班。四、完整实操案例阻塞队列importjava.util.ArrayList;importjava.util.List;importjava.util.concurrent.TimeUnit;importjava.util.concurrent.locks.Condition;importjava.util.concurrent.locks.ReentrantLock;publicclassBlockingQueueT{privateReentrantLocklocknewReentrantLock();privateConditionfulllock.newCondition();//生产者等待区队列满时privateConditionemptylock.newCondition();//消费者等待区队列空时privateListTqueue;privateintmaxSize;publicBlockingQueue(intmaxSize){this.maxSizemaxSize;queuenewArrayList(maxSize);}//阻塞添加publicvoidput(Ttask){lock.lock();try{while(queue.size()maxSize){System.out.println(Thread.currentThread().getName():队列已满进入等待…);full.await();//满了生产者在 full 中等待}queue.add(task);empty.signal();//放进任务了叫醒一个消费者}catch(InterruptedExceptione){thrownewRuntimeException(e);}finally{lock.unlock();}}//阻塞获取publicTtake(){lock.lock();try{while(queue.isEmpty()){System.out.println(Thread.currentThread().getName():队列已空进入等待…);empty.await();//空了消费者在 empty 中等待}Ttaskqueue.removeFirst();full.signal();//取走了任务叫醒一个生产者returntask;}catch(InterruptedExceptione){thrownewRuntimeException(e);}finally{lock.unlock();}}//带超时的阻塞获取publicTpoll(longtimeOut,TimeUnitunit){lock.lock();try{longnanosunit.toNanos(timeOut);//统一转成纳秒while(queue.isEmpty()){System.out.println(Thread.currentThread().getName():队列已空进入等待…);if(nanos0){returnnull;//等超时了直接返回}nanosempty.awaitNanos(nanos);//等待指定时间返回剩余时间}Ttaskqueue.removeFirst();full.signal();returntask;}catch(InterruptedExceptione){thrownewRuntimeException(e);}finally{lock.unlock();}}}线程池仅创建临时线程importjava.util.ArrayList;importjava.util.List;importjava.util.concurrent.TimeUnit;publicclassThreadPool{privateBlockingQueueRunnableblockingQueue;privateintcoreSize;privatelongtimeOut;privateTimeUnitunit;privateListWorkerThreadworkersnewArrayList();//正在干活的线程publicThreadPool(BlockingQueueRunnableblockingQueue,intcoreSize,longtimeOut,TimeUnitunit){this.blockingQueueblockingQueue;this.coreSizecoreSize;this.timeOuttimeOut;this.unitunit;}//提交任务publicvoidexecute(Runnabletask){synchronized(workers){if(workers.size()coreSize){//线程数还没满直接新建一个临时线程并开始干活WorkerThreadworkernewWorkerThread(task);workers.add(worker);worker.start();}else{//线程满了任务丢进队列排队blockingQueue.put(task);}}}//工作线程classWorkerThreadextendsThread{privateRunnabletask;publicWorkerThread(Runnabletask){this.tasktask;}Overridepublicvoidrun(){//如果有初始任务或者能从队列里拿到新任务就一直干while(task!null||(taskblockingQueue.poll(timeOut,unit))!null){try{task.run();//执行任务}catch(Exceptione){e.printStackTrace();//任务报错不能把线程搞死}finally{tasknull;//干完清空准备拿下一个}}//拿不到任务等待超时了把自己从线程池里移除synchronized(workers){workers.remove(this);System.out.println(Thread.currentThread().getName():等待超时退出);}}}publicstaticvoidmain(String[]args){//创建一个线程池线程2个等待超时2秒任务队列容量5BlockingQueueRunnablequeuenewBlockingQueue(5);ThreadPoolpoolnewThreadPool(queue,2,2,TimeUnit.SECONDS);//连续提交6个任务for(inti1;i6;i){intji;pool.execute(newRunnable(){Overridepublicvoidrun(){System.out.println(Thread.currentThread().getName() 正在执行任务 j);try{//故意让任务执行得慢一点1秒方便观察线程复用Thread.sleep(1000);}catch(InterruptedExceptione){e.printStackTrace();}}});}}}大概会看到这样的输出Thread-0 正在执行任务 1 Thread-1 正在执行任务 2 Thread-1 正在执行任务 3 Thread-0 正在执行任务 4 Thread-1 正在执行任务 6 Thread-0 正在执行任务 5 Thread-0:队列已空进入等待… Thread-1:队列已空进入等待… Thread-1:队列已空进入等待… Thread-0:队列已空进入等待… Thread-1:等待超时退出 Thread-0:等待超时退出五、个人收获总结这次最大的收获就是把以前学习的线程相关知识都串连了起来从独立线程到生产消费模型再到线程池……用新旧知识的结合我一步步解决了同时调用方法的数据紊乱问题实现了线程的复用并用阻塞队列和线程池控制与管理线程们。这次练习主要拓展了锁ReentrantLock和缓冲区Condition的应用制作了一个更贴近官方版的线程池与阻塞队列并实现了相关基础功能。令我对线程的理解更加深刻了。拓展阅读Java多线程基础笔记独立线程Java多线程基础笔记生产者-消费者模型Java多线程基础笔记线程池