多线程高并发编程(12) -- 阻塞算法实现ArrayBlockingQueue源码分析

源码仓库Gitee · GitHub一.前言前文探究了非阻塞算法的实现ConcurrentLinkedQueue安全队列也说明了阻塞算法实现的两种方式使用一把锁出队和入队同一把锁ArrayBlockingQueue和两把锁出队和入队各一把锁LinkedBlockingQueue来实现今天来探究下ArrayBlockingQueue。ArrayBlockingQueue是一个阻塞队列底层使用数组结构实现按照先进先出FIFO的原则对元素进行排序。ArrayBlockingQueue是一个线程安全的集合通过ReentrantLock锁来实现在并发情况下可以保证数据的一致性。此外ArrayBlockingQueue的容量是有限的数组的大小在初始化时就固定了不会随着队列元素的增加而出现扩容的情况也就是说ArrayBlockingQueue是一个“有界缓存区”。从下图可以看出ArrayBlockingQueue是使用一个数组存储元素的当向队列插入元素时首先会插入到数组下标索引为6的位置再有新元素进来时插入到索引为7的位置依次类推如果满了就不会再插入。当元素出队时先移除索引为2的元素3与入队一样依次类推移除索引3、4、5...上的元素。这也形成了“先进先出”。二.源码解析构造方法public class ArrayBlockingQueueE extends AbstractQueueE implements BlockingQueueE, java.io.Serializable { //队列实现数组 final Object[] items; //当读取元素时数组的下标(下一个被取出元素的索引) int takeIndex; //添加元素时数组的下标 下一个被添加元素的索引 int putIndex; //队列中元素个数 int count; //可重入锁 final ReentrantLock lock; //入队操作时是否让线程等待 private final Condition notEmpty; //出队操作时是否让线程等待 private final Condition notFull; /** * 初始化队列容量构造由于公平锁会降低队列的性能因而使用非公平锁(默认)。 */ public ArrayBlockingQueue(int capacity) { this(capacity, false); } //带初始容量大小和公平锁队列(公平锁通过ReentrantLock实现) public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity 0) throw new IllegalArgumentException(); this.items new Object[capacity]; lock new ReentrantLock(fair); notEmpty lock.newCondition(); notFull lock.newCondition(); } }在多线程中默认不保证线程公平的访问队列在ArrayBlockingQueue中为了保证数据的安全使用了ReentrantLock锁。由于锁的引入导致了线程之间的竞争。当有一个线程获取到锁时其余线程处于等待状态。当锁被释放时所有等待线程为夺锁而竞争锁有公平锁和非公平锁公平锁等待的线程在获取锁而竞争时按照等待的先后顺序FIFO进行获取操作;公平锁可以应用在比如并发下的日志输出队列中保证了日志输出的顺序完整性优点等待锁的线程不会饿死和非公平锁相比在获得锁和保证锁分配的均衡性差异较小缺点使用公平锁的程序在多线程访问时表现为很低的吞吐量即速度很慢等待队列中除第一个线程以外的所有线程都会阻塞CPU唤醒阻塞线程的开销比非公平锁的大公平锁不能保证线程调度的公平性因此使用公平锁的众多线程中的一员可能获得多倍的成功机会这种情况发生在其他活动线程没有被处理并且目前并未持有锁时【ReentrantLock源码对公平锁的定义】Note however, that fairness of locks does not guarantee fairness of thread scheduling. Thus, one of many threads using a fair lock may obtain it multiple times in succession while other active threads are not progressing and not currently holding the lock.上面这句话有重入锁的概念一个线程可以在已经获取锁的情况下再次进入获取到锁不需要竞争同时如果一个线程获取到了锁然后释放在其他线程来获取之前再次是可以获取到锁的。A: Request Lock - Release Lock - Request Lock Again (Succeeds) B: Request Lock (Denied)... ----------------------- Time ---------------------------------非公平锁在获取锁时无论是先等待还是后等待的线程均有可能获取到锁。即根据抢占机制是随机获取锁的和公平锁不一样的是先来的不一定能获取到锁有可能一直拿不到锁这样会造成“饥饿”现象优点非公平锁性能高于公平锁性能。首先在恢复一个被挂起的线程与该线程真正运行之间存在着严重的延迟而且非公平锁更能充分的利用CPU的时间片尽量减少CPU空闲的状态时间即可以减少唤起线程的开销整体的吞吐效率高因为线程有几率不阻塞直接获取到锁CPU不必唤醒其他所有线程缺点处于等待队列中的线程可能会饿死或者等很久才会获得锁产生“饥饿”的原因高优先级吞噬所有低优先级的CPU时间片优先级越高就会获得越高的CPU执行机会 --- 使用默认的优先级线程被永久阻塞在一个等待进入同步块synchronized的状态长时间执行 同时synchronized并不保障等待线程的顺序锁释放后随机竞争由OS调度这会存在一个可能是某个线程总是抢锁抢不到导致一直等待状态 --- 避免持有锁的线程长时间执行、使用显示lock来代替synchronizedsynchronized(obj) { while (true) { // .... infinite loop }等待的线程永远不被唤醒如果多个线程处在wait方法执行上而对其调用notify方法不会保证哪一个线程会获得唤醒唤醒是无序的跟VM/OS调度有关甚至底层是随机选取一个或是队列中的第一个任何线程都有可能处于继续等待的状态因此存在这样一个风险即一个等待线程从来得不到唤醒因为其他等待线程总是能被获得唤醒 --- 使用显示lock来代替synchronized比如ReentrantLock在公平锁中如果有另一个线程持有锁或者有其他线程在等待队列中等待这个锁那么新发出的请求的线程将被放入到队列中非公平锁中, 根据抢占机制拥有锁的线程在释放锁资源的时候, 新发出请求的线程可以和等待队列中的第一个线程竞争锁资源, 新线程竞争失败才放入队列中但是已经进入等待队列的线程, 依然是按照先进先出的顺序获取锁资源入队有阻塞式和非阻塞式阻塞式当队列中的元素已满时则会将此线程停止让其处于等待状态直到队列中有空余位置产生public void put(E e) throws InterruptedException { checkNotNull(e); final ReentrantLock lock this.lock; lock.lockInterruptibly();//获取锁 try { //队列中元素 数组长度(队列满了),则线程等待 while (count items.length) notFull.await(); enqueue(e);//元素加入队列 } finally { lock.unlock();//释放锁 } }lockInterruptibly如果当前线程未被中断则获取锁。如果该锁没有被另一个线程保持则获取该锁并立即返回将锁的保持计数设置为 1。如果当前线程已经保持此锁则将保持计数加 1并且该方法立即返回。如果锁被另一个线程保持则出于线程调度目的禁用当前线程并且在发生以下两种情况之一以前该线程将一直处于休眠状态1锁由当前线程获得2其他某个线程中断当前线程非阻塞式当队列中的元素已满时并不会阻塞此线程的操作而是让其返回又或者是抛出异常public boolean add(E e) { return super.add(e);// AbstractQueue.add } public boolean add(E e) { if (offer(e))//调用实现接口 return true; else throw new IllegalStateException(Queue full); } public boolean offer(E e) { checkNotNull(e);//检测是否有空指针异常 final ReentrantLock lock this.lock;//获得锁对象 lock.lock();//加锁 try { //如果队列满了返回false if (count items.length) return false; else { //元素加入队列 enqueue(e); return true; } } finally { lock.unlock();//释放锁 } } private void enqueue(E x) { // assert lock.getHoldCount() 1; // assert items[putIndex] null; //获得数组 final Object[] items this.items; //槽位填充元素 items[putIndex] x; //获得下一个被添加元素的索引如果值等于数组长度表示到达尾部了需要从头开始填充 if (putIndex items.length) putIndex 0; count;//数量1 notEmpty.signal();//唤醒出队上的等待线程表示有元素可以消费了 }enqueue中putIndex items.lengthputIndex0这是因为当前队列执行元素出队时总是从队列头部获取而添加元素的索引从队列尾部获取所以当队列索引从0开始与数组长度相等时下次我们就需要从数组头部开始添加了阻塞式和非阻塞式的结合offer(E e, long timeout, TimeUnit unit)向队列尾部添加元素可以设置线程等待时间如果超过指定时间队列还是满的则返回falsepublic boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { checkNotNull(e);//检测是否为空 long nanos unit.toNanos(timeout);//转换成超时时间阀值 final ReentrantLock lock this.lock; lock.lockInterruptibly();//加锁 try { //队列是否满了的判断 while (count items.length) { if (nanos 0)//等待超时结束返回false return false; nanos notFull.awaitNanos(nanos);//队列满了等待出队有空位填充 } enqueue(e);//加入队列中 return true; } finally { lock.unlock();//释放锁 } }出队同样有阻塞式和非阻塞式阻塞式当队列中的元素已空时则会将此线程停止让其处于等待状态直到队列中有元素插入public E take() throws InterruptedException { final ReentrantLock lock this.lock; lock.lockInterruptibly(); try { //队列为空进行等待 while (count 0) notEmpty.await(); return dequeue();//返回出队元素 } finally { lock.unlock(); } }非阻塞式当队列中的元素已满时并不会阻塞此线程的操作而是让其返回null或元素【里面的迭代器比较复杂留待下文探究】public E poll() { final ReentrantLock lock this.lock; lock.lock(); try { //队列为空返回null否则返回元素 return (count 0) ? null : dequeue(); } finally { lock.unlock(); } } private E dequeue() { // assert lock.getHoldCount() 1; // assert items[takeIndex] ! null; final Object[] items this.items;//获得队列 SuppressWarnings(unchecked) E x (E) items[takeIndex];//获得出队元素 items[takeIndex] null;//出队槽位元素置为null //下一个被取出元素的索引1如果值等于长度表示后面没有元素了需要从头开始取出 if (takeIndex items.length) takeIndex 0; count--;//数量-1 if (itrs ! null)//迭代器不为空 itrs.elementDequeued();//同时更新迭代器中的元素数据 notFull.signal();//唤醒入队线程 return x;//返回出队元素 }阻塞式和非阻塞式的结合poll(long timeout, TimeUnit unit)出队获取元素可以设置线程等待时间如果超过指定时间队列还是空的则返回nullpublic E poll(long timeout, TimeUnit unit) throws InterruptedException { long nanos unit.toNanos(timeout);//转换成超时时间阀值 final ReentrantLock lock this.lock; lock.lockInterruptibly();//加锁 try { while (count 0) {//队列空了等待 if (nanos 0)//超时了返回null return null; nanos notEmpty.awaitNanos(nanos);//等待入队填充元素 } return dequeue();//返回出队元素 } finally { lock.unlock();//释放锁 } }移除元素removepublic boolean remove(Object o) { //要移除的元素为空返回false if (o null) return false; //获得队列数组 final Object[] items this.items; final ReentrantLock lock this.lock; lock.lock();//加锁 try { //队列有元素 if (count 0) { final int putIndex this.putIndex;//获得下一个被添加元素的索引 int i takeIndex;//下一个被取出元素的索引 do { if (o.equals(items[i])) {//从takeIndex下标开始找到要被删除的元素 removeAt(i);//移除 return true; } if (i items.length)//下一个被取出元素的索引1并判断是否等于队列长度如果是表示需要从头开始遍历 i 0; } while (i ! putIndex);//继续查找直到找到最后一个元素 } return false; } finally { lock.unlock();//解锁 } } /** * 根据下标移除元素那么会分成两种情况一个是移除的是队首元素一个是移除的是非队首元素,移除队首元素就相当于出队操作 * 移除非队首元素那么中间就有空位了后面元素需要依次补上然后如果是队尾元素那么putIndex也就是插入操作的下标也就需要跟着移动。 */ void removeAt(final int removeIndex) { // assert lock.getHoldCount() 1; // assert items[removeIndex] ! null; // assert removeIndex 0 removeIndex items.length; final Object[] items this.items;//获得队列 if (removeIndex takeIndex) {//移除的是队首元素 // removing front item; just advance items[takeIndex] null;//队首置为null if (takeIndex items.length)//下一个被取出元素的索引1并判断是否等于队列长度 takeIndex 0; count--;//数量-1 if (itrs ! null)//迭代器不为空 itrs.elementDequeued();//更新迭代器元素 } else {//移除的不是队首元素而是中间元素 // an interior remove // slide over all others up through putIndex. final int putIndex this.putIndex;//下一个被添加元素的索引 for (int i removeIndex;;) {//对队列进行遍历因为是队列中间的值被移除了所有后面的元素都要挨个迁移 int next i 1;//获取移除元素的下一个坐标 if (next items.length)//判断是否等于队列长度 next 0; if (next ! putIndex) {//获取移除元素的下一个坐标下一个被添加元素的索引表示移除元素的索引后面有值 items[i] items[next];//当前要移除的元素置为后面的元素即对后面的元素往前迁移覆盖要移除的元素 i next;//下一个迁移的索引 } else {//移除的元素是最后一个后面没有值了 items[i] null;//移除元素直接置为null this.putIndex i;//更新下一个被添加元素的索引 break;//结束 } } count--;//数量-1 if (itrs ! null)//迭代器不为空 itrs.removedAt(removeIndex);//更新迭代器元素 } notFull.signal();//唤醒入队线程可以添加元素了 }清空元素clear用于清空ArrayBlockingQueue并且会释放所有等待notFull条件的线程存放元素的线程public void clear() { final Object[] items this.items;//获得队列 final ReentrantLock lock this.lock; lock.lock(); try { int k count;//获取元素数量 if (k 0) {//有元素表示队列不为空 final int putIndex this.putIndex;//下一个被添加元素的索引 int i takeIndex;//下一个被取出元素的索引 do { items[i] null;//对每个有元素的槽位置为null if (i items.length) i 0; } while (i ! putIndex);//从有元素的第一个槽位开始遍历直到槽位元素为null takeIndex putIndex;//更新取出和添加的索引 count 0;//数量更新为0 if (itrs ! null)//迭代器不为空 itrs.queueIsEmpty();//更新迭代器为空 //若有等待notFull条件的线程则逐一唤醒 for (; k 0 lock.hasWaiters(notFull); k--) notFull.signal();//唤醒入队线程可以添加元素了 } } finally { lock.unlock(); } }offer(E e, long timeout, TimeUnit unit)和poll(long timeout, TimeUnit unit)里面有awaitNanos下面探讨该功能实现对当前线程或等待的入/出队线程进行挂起如果有入/出队操作进行了唤醒出/入队操作则acquireQueued自旋获取到锁然后出/入队中的ReentrantLock是重入锁可以重入获取到锁进行出/入队操作AbstractQueuedSynchronizer //进行超时控制 public final long awaitNanos(long nanosTimeout) throws InterruptedException { //如果当前线程中断了抛出中断异常 if (Thread.interrupted()) throw new InterruptedException(); //当前线程加入到Condition队列中 Node node addConditionWaiter(); //锁释放是否成功释放当前线程的lock从AQS的队列中移出 int savedState fullyRelease(node); //到达等待时间点 final long deadline System.nanoTime() nanosTimeout; //中断标识 int interruptMode 0; //当前节点是否在同步队列中否表示不在进入挂起判断操作如果已经在Sync队列中则退出循环 //那什么时候会把当前线程又加入到Sync队列中呢当然是调用signal方法的时候因为这里需要唤醒之前调用await方法的线程唤醒之后进行下面的获取锁等操作 while (!isOnSyncQueue(node)) { //如果超时了将线程挂起然后停止遍历 if (nanosTimeout 0L) { transferAfterCancelledWait(node); break; } //如果等待时间间隔超过了1000继续挂起 if (nanosTimeout spinForTimeoutThreshold) LockSupport.parkNanos(this, nanosTimeout); //线程中断了停止遍历 if ((interruptMode checkInterruptWhileWaiting(node)) ! 0) break; //获得剩余的等待时间间隔 nanosTimeout deadline - System.nanoTime(); } //结束挂起acquireQueued自旋对当前线程的队列出队进行获取锁并返回线程是否中断 //如果线程被中断并且中断的方式不是抛出异常则设置中断后续的处理方式设置为REINTERRUPT if (acquireQueued(node, savedState) interruptMode ! THROW_IE) interruptMode REINTERRUPT;//中断标识更新为退出等待时重新中断 if (node.nextWaiter ! null)//当前节点后面还有节点多并发操作了 unlinkCancelledWaiters();//从头到尾遍历Condition队列移除被cancel的节点 //如果线程已经被中断则根据之前获取的interruptMode的值来判断是继续中断还是抛出异常 if (interruptMode ! 0) reportInterruptAfterWait(interruptMode); return deadline - System.nanoTime();//返回剩余等待时间 }drainTo可以一次性获取队列中所有的元素它减少了锁定队列的次数使用得当在某些场景下对性能有不错的提升//最多从此队列中移除给定数量的可用元素并将这些元素添加到给定collection中 public int drainTo(Collection? super E c) { return drainTo(c, Integer.MAX_VALUE); } public int drainTo(Collection? super E c, int maxElements) { checkNotNull(c);//检查是否为空 if (c this)//如果集合类型相同抛出参数异常 throw new IllegalArgumentException(); if (maxElements 0)//如果给定移除数量小于0返回0表示不做移除操作 return 0; final Object[] items this.items;//获得队列 final ReentrantLock lock this.lock; lock.lock();//加锁 try { int n Math.min(maxElements, count);//获得元素的最小数量 int take takeIndex;//下一个被取出元素的索引 int i 0; try { while (i n) {//遍历移除和添加 SuppressWarnings(unchecked) E x (E) items[take];//获得移除元素 c.add(x);//元素添加到直到集合中 items[take] null;//元素原先队列位置置为null if (take items.length)//如果取出索引到达尾部从头开始遍历取出 take 0; i;//移除的数量1如果达到了移除的最小数量结束遍历 } return n;//返回一共移除并添加了多少个元素 } finally { // Restore invariants even if c.add() threw if (i 0) {//如果有移除操作 count - i;//队列元素数量-i takeIndex take;//重置下一个被取出元素的索引 if (itrs ! null) {//迭代器不为空 if (count 0)//队列空了 itrs.queueIsEmpty();//迭代器清空 else if (i take)//说明take中间变成0了通知itr itrs.takeIndexWrapped(); } //唤醒在因为队列满而等待的入队线程最多唤醒i个避免线程被唤醒了因为队列又满了而阻塞 for (; i 0 lock.hasWaiters(notFull); i--) notFull.signal(); } } } finally { lock.unlock(); } }三.Logback 框架中异步日志打印中ArrayBlockingQueue的使用在高并发并且响应时间要求比较小的系统中同步打日志已经满足不了需求了这是因为打日志本身是需要同步写磁盘的会造成 响应时间 增加如下图同步日志打印模型为异步模型是业务线程把要打印的日志任务写入一个队列后直接返回然后使用一个线程专门负责从队列中获取日志任务写入磁盘其模型具体如下图如图可知其实 logback 的异步日志模型是一个多生产者单消费者模型通过使用队列把同步日志打印转换为了异步业务线程调用异步 appender 只需要把日志任务放入日志队列日志线程则负责使用同步的 appender 进行具体的日志打印到磁盘接下来看看异步日志打印具体实现要把同步日志打印改为异步需要修改 logback 的 xml 配置文件appender namePROJECT classch.qos.logback.core.FileAppender fileproject.log/file encodingUTF-8/encoding appendtrue/append rollingPolicy classch.qos.logback.core.rolling.TimeBasedRollingPolicy !-- daily rollover -- fileNamePatternproject.log.%d{yyyy-MM-dd}/fileNamePattern !-- keep 7 days worth of history -- maxHistory7/maxHistory /rollingPolicy layout classch.qos.logback.classic.PatternLayout pattern ![CDATA[%n%-4r [%d{yyyy-MM-dd HH:mm:ss}] %X{productionMode} - %X{method} %X{requestURIWithQueryString} [ip%X{remoteAddr}, ref%X{referrer}, ua%X{userAgent}, sid%X{cookie.JSESSIONID}]%n %-5level %logger{35} - %m%n]] /pattern /layout /appender appender nameasyncProject classch.qos.logback.classic.AsyncAppender discardingThreshold0/discardingThreshold queueSize1024/queueSize neverBlocktrue/neverBlock appender-ref refPROJECT / /appender logger namePROJECT_LOGGER additivityfalse level valueWARN / appender-ref refasyncProject / /logger从上面可知 AsyncAppender 是实现异步日志的关键下面探究它的原理如上图可知 AsyncAppender 继承自 AsyncAppenderBase其中后者具体实现了异步日志模型的主要功能前者只是重写了其中的一些方法。另外从类图可知 logback 中的异步日志队列是一个阻塞队列, 后面会知道其实是一个有界阻塞队列 ArrayBlockingQueue, 其中 queueSize 是有界队列的元素个数默认为 256worker则是工作线程也就是异步打印日志的消费者线程aai则是一个appender的装饰器里边存放的同步日志的appender其中appenderCount记录aai里边附加的同步appender的个数这个和配置文件相对应一个异步的appender对应一个同步的appenderneverBlock用来指示当同步队列已满时是否阻塞打印日志线程如果配置neverBlocktrue当队列满了之后后面阻塞的线程想要输出的消息就直接被丢弃从而线程不会阻塞discardingThreshold是一个阈值当日志队列里边的空闲元素个数小于该值时新来的某些级别的日志就会直接被丢弃。接下来看下何时创建的日志队列以及何时启动的消费线程这需要看下 AsyncAppenderBase 的 start 方法该方法是在解析完毕配置 AsyncAppenderBase 的 xml 的节点元素后被调用 public void start() { if (isStarted()) return; if (appenderCount 0) { addError(No attached appenders found.); return; } if (queueSize 1) { addError(Invalid queue size [ queueSize ]); return; } // 创建一个ArrayBlockingQueue阻塞队列queueSize默认为256创建阻塞队列的原因是防止生产者过多造成队列中元素过多产生OOM异常 blockingQueue new ArrayBlockingQueueE(queueSize); // 如果discardingThreshold未定义的话默认为queueSize的1/5 if (discardingThreshold UNDEFINED) discardingThreshold queueSize / 5; addInfo(Setting discardingThreshold to discardingThreshold); // 将工作线程设置为守护线程即当jvm停止时即使队列中有未处理的元素也不会在进行处理 worker.setDaemon(true); // 为线程设置name便于调试 worker.setName(AsyncAppender-Worker- getName()); // make sure this instance is marked as started before staring the worker Thread // 启动线程 super.start(); worker.start(); }logback 使用的队列是有界队列 ArrayBlockingQueue之所以使用有界队列是考虑到内存溢出问题在高并发下写日志的 qps 会很高如果设置为无界队列队列本身会占用很大内存很可能会造成 内存溢出。这里消费日志队列的 worker 线程被设置为了守护线程意味着当主线程运行结束并且当前没有用户线程时候该 worker 线程会随着 JVM 的退出而终止而不管日志队列里面是否还有日志任务未被处理。另外这里设置了线程的名称是个很好的习惯因为这在查找问题的时候很有帮助根据线程名字就可以定位到是哪个线程。既然是有界队列那么肯定需要考虑如果队列满了该如何处置是丢弃老的日志任务还是阻塞日志打印线程直到队列有空余元素那?下面看append 方法protected void append(E eventObject) { // 判断队列中的元素数量是否小于discardingThreshold如果小于的话并且日志等级小于info的话则直接丢弃这些日志任务 if (isQueueBelowDiscardingThreshold() isDiscardable(eventObject)) { return; } preprocess(eventObject); // 日志入队 put(eventObject); } private boolean isQueueBelowDiscardingThreshold() { return (blockingQueue.remainingCapacity() discardingThreshold); } // 子类重写的方法 判断日志等级 protected boolean isDiscardable(ILoggingEvent event) { Level level event.getLevel(); return level.toInt() Level.INFO_INT; }日志入队put从下面可知如果 neverBlock 设置为 false默认为 false则会调用阻塞队列的 put 方法而 put 是阻塞的也就是说如果当前队列满了如果再企图调用 put 方法向队列放入一个元素则调用线程会被阻塞直到队列有空余空间。这里有必要提下其中blockingQueue.put(eventObject)当日志队列满了的时候 put 方法会调用 await() 方法阻塞当前线程如果其它线程中断了该线程那么该线程会抛出 InterruptedException 异常那么当前的日志任务就会被丢弃了。如果 neverBlock 设置为了 true 则会调用阻塞队列的 offer 方法而该方法是非阻塞的如果当前队列满了则会直接返回也就是丢弃当前日志任务。private void put(E eventObject) { // 判断是否阻塞默认为false则会调用阻塞队列的put方法 if (neverBlock) { blockingQueue.offer(eventObject); } else { putUninterruptibly(eventObject); } } // 可中断的阻塞put方法 private void putUninterruptibly(E eventObject) { boolean interrupted false; try { while (true) { try { blockingQueue.put(eventObject); break; } catch (InterruptedException e) { interrupted true; } } } finally { if (interrupted) { Thread.currentThread().interrupt(); } } }最后看下 addAppender 方法可以看出一个异步的appender只能绑定一个同步appender这个appender会被放入AppenderAttachableImpl的appenderList列表里边public void addAppender(AppenderE newAppender) { if (appenderCount 0) { appenderCount; addInfo(Attaching appender named [ newAppender.getName() ] to AsyncAppender.); aai.addAppender(newAppender); } else { addWarn(One and only one appender may be attached to AsyncAppender.); addWarn(Ignoring additional appender named [ newAppender.getName() ]); } }通过上面我们已经分析完了日志生产线程放入日志任务到日志队列的实现下面一起来看下消费线程是如何从队列里面消费日志任务并写入磁盘的由于消费线程是一个线程那就从 worker 的 run 方法看起消费者将日志写入磁盘的线程方法class Worker extends Thread { public void run() { AsyncAppenderBaseE parent AsyncAppenderBase.this; AppenderAttachableImplE aai parent.aai; // loop while the parent is started 一直循环知道线程被中断 while (parent.isStarted()) { try {// 从阻塞队列中获取元素交由给同步的appender将日志打印到磁盘 E e parent.blockingQueue.take(); aai.appendLoopOnAppenders(e); } catch (InterruptedException ie) { break; } } addInfo(Worker thread will flush remaining events before exiting. ); //执行到这里说明该线程被中断则把队列里边的剩余日志任务刷新到磁盘 for (E e : parent.blockingQueue) { aai.appendLoopOnAppenders(e); parent.blockingQueue.remove(e); } aai.detachAndStopAllAppenders(); } }try逻辑中从日志队列使用 take 方法获取一个日志任务如果当前队列为空则当前线程会阻塞到 take 方法直到队列不为空才返回获取到日志任务后会调用 AppenderAttachableImpl 的 aai.appendLoopOnAppenders 方法该方法会循环调用通过 addAppender 注入的同步日志 appener 具体实现日志打印到磁盘的任务。四.参考公平锁的使用场景https://stackoverflow.com/questions/26455578/when-to-use-fairness-mode-in-java-concurrency公平锁和非公平锁的区别的提问https://segmentfault.com/q/1010000006439146公平锁不能保证线程调度的公平性https://stackoverflow.com/questions/60903107/understanding-fair-reentrantlock-in-javalogback异步日志打印中的ArrayBlockingQueue的使用https://my.oschina.net/u/4410397/blog/3428573/print欢迎任何形式的转载但请务必注明出处。如果觉得文章还行希望推荐关注下谢谢限于本人水平如果文章和代码有表述不当之处还请不吝赐教。