# Nicksxs's Blog

What hurts more, the pain of hard work or the pain of regret?

0%

## AQS篇二 之 Condition 浅析笔记

发表于 2021-02-21 分类于 Java ， 并发 阅读次数： Disqus：

Condition也是 AQS 中很重要的一块内容，可以先看段示例代码，这段代码应该来自于Doug Lea大大，可以在 javadoc 中的 condition 部分找到，其实大大原来写过基于 `synchronized` 实现的，后面我也贴下代码

    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;

    class BoundedBuffer {
        final Lock lock = new ReentrantLock();
        // condition 依赖于 lock 来产生
        final Condition notFull = lock.newCondition();
        final Condition notEmpty = lock.newCondition();

        // 对象池子，put 跟 take 的就是这里的
        final Object[] items = new Object[100];
        int putptr, takeptr, count;

        // 生产
        public void put(Object x) throws InterruptedException {
            // 这里也说明了，需要先拥有锁
            lock.lock();
            try {
                while (count == items.length)
                    notFull.await();  // 队列已满，等待，直到 not full 才能继续生产
                items[putptr] = x;
                if (++putptr == items.length) putptr = 0;
                ++count;
                notEmpty.signal(); // 生产成功，队列已经 not empty 了，发个通知出去
            } finally {
                lock.unlock();
            }
        }

        // 消费
        public Object take() throws InterruptedException {
            lock.lock();
            try {
                while (count == 0)
                    notEmpty.await(); // 队列为空，等待，直到队列 not empty，才能继续消费
                Object x = items[takeptr];
                if (++takeptr == items.length) takeptr = 0;
                --count;
                notFull.signal(); // 被我消费掉一个，队列 not full 了，发个通知出去
                return x;
            } finally {
                lock.unlock();
            }
        }
    }

介绍下 Condition 的结构

    public class ConditionObject implements Condition, java.io.Serializable {
            private static final long serialVersionUID = 1173984872572414699L;
            /** First node of condition queue. */
            private transient Node firstWaiter;
            /** Last node of condition queue. */
            private transient Node lastWaiter;

主要的就这么点，而且也复用了 AQS 阻塞队列或者大大叫 `lock queue`中同样的 Node 节点，只不过它没有使用其中的双向队列，也就是prev 和 next，而是在 Node 中的 nextWaiter，所以只是个单向的队列，没使用 next 其实还有个用处，后面会提到，看下结构的示意图
然后主要是看两个方法，`await` 和 `signal`,
先来看下 await

    /**
     * Implements interruptible condition wait.
     * <ol>
     * <li> If current thread is interrupted, throw InterruptedException.
     * <li> Save lock state returned by {@link #getState}.
     * <li> Invoke {@link #release} with saved state as argument,
     *      throwing IllegalMonitorStateException if it fails.
     * <li> Block until signalled or interrupted.
     * <li> Reacquire by invoking specialized version of
     *      {@link #acquire} with saved state as argument.
     * <li> If interrupted while blocked in step 4, throw InterruptedException.
     * </ol>
     */
    public final void await() throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();

        // 将当前节点包装成一个  condition waiter node 节点
        Node node = addConditionWaiter();

        // 完全释放占有的锁，这里需要是占有锁的线程
        int savedState = fullyRelease(node);
        int interruptMode = 0;

        // 判断下是否在阻塞队列中，因为有可能被其他节点从等待队列移动到阻塞队列
        while (!isOnSyncQueue(node)) {
            // park等待，等待被唤醒
            LockSupport.park(this);
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }

        // 被唤醒后进入阻塞队列，等待获取锁，这里继续用了fullyRelease返回的 state
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
    }

添加条件队列节点

    /**
     * Adds a new waiter to wait queue.
     * @return its new wait node
     */
    private Node addConditionWaiter() {
        Node t = lastWaiter;
        // If lastWaiter is cancelled, clean out.
        // 如果节点已经不是 CONDITION 状态了，表示已经取消了
        if (t != null && t.waitStatus != Node.CONDITION) {
            // 把等待队列中取消的节点清理出去
            unlinkCancelledWaiters();
            t = lastWaiter;
        }
        // 把当前线程包装成waitStatus=CONDITION 的节点
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        // 没有 lastWaiter 节点，直接是 firstWaiter
        if (t == null)
            firstWaiter = node;
        else
        // 不然就接在 lastWaiter 后面
            t.nextWaiter = node;
        // 当前节点就会变成新的 lastWaiter
        lastWaiter = node;
        return node;
    }

清理取消的节点

    /**
     * Unlinks cancelled waiter nodes from condition queue.
     * Called only while holding lock. This is called when
     * cancellation occurred during condition wait, and upon
     * insertion of a new waiter when lastWaiter is seen to have
     * been cancelled. This method is needed to avoid garbage
     * retention in the absence of signals. So even though it may
     * require a full traversal, it comes into play only when
     * timeouts or cancellations occur in the absence of
     * signals. It traverses all nodes rather than stopping at a
     * particular target to unlink all pointers to garbage nodes
     * without requiring many re-traversals during cancellation
     * storms.
     */
    private void unlinkCancelledWaiters() {
        Node t = firstWaiter;
        Node trail = null;
        // 循环遍历单向链表的节点，如果状态不是 CONDITION 就清出去
        while (t != null) {
            Node next = t.nextWaiter;
            // 循环链表操作，清掉取消的节点
            if (t.waitStatus != Node.CONDITION) {
                t.nextWaiter = null;
                if (trail == null)
                    firstWaiter = next;
                else
                    trail.nextWaiter = next;
                if (next == null)
                    lastWaiter = trail;
            }
            else
                trail = t;
            t = next;
        }
    }

完全释放锁

    /**
     * Invokes release with current state value; returns saved state.
     * Cancels node and throws exception on failure.
     * @param node the condition node for this wait
     * @return previous sync state
     */
    final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            // 获取下当前的 state 值，因为是可重入的，所以这个值要保存下来
            int savedState = getState();
            // 这里还包含比较多操作，不过跟前面分析 AQS 的释放比较类似，不深入了
            if (release(savedState)) {
                failed = false;
                // 返回这个值
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }

判断是否在阻塞队列中

    /**
     * Returns true if a node, always one that was initially placed on
     * a condition queue, is now waiting to reacquire on sync queue.
     * @param node the node
     * @return true if is reacquiring
     */
    final boolean isOnSyncQueue(Node node) {
        // 如果waitStatus 是 CONDITION 或者没有 prev 前置节点肯定就不在
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        // 这里就是我前面提到的 next 的作用
        if (node.next != null) // If has successor, it must be on queue
            return true;
        // 从 tail 开始找，是否在阻塞队列中
        /*
         * node.prev can be non-null, but not yet on queue because
         * the CAS to place it on queue can fail. So we have to
         * traverse from tail to make sure it actually made it.  It
         * will always be near the tail in calls to this method, and
         * unless the CAS failed (which is unlikely), it will be
         * there, so we hardly ever traverse much.
         */
        return findNodeFromTail(node);
    }
    /**
     * Returns true if node is on sync queue by searching backwards from tail.
     * Called only when needed by isOnSyncQueue.
     * @return true if present
     */
    private boolean findNodeFromTail(Node node) {
        Node t = tail;
        // 从 tail 开始，从后往前找
        for (;;) {
            if (t == node)
                return true;
            if (t == null)
                return false;
            t = t.prev;
        }
    }

await 的逻辑差不多就是这样子，主要的就是把自己包成一个 Node 节点，waitStatus 的状态是 CONDITION，挂在等待队列的最后，然后完全释放锁，park 等待

## signal

    /**
     * Moves the longest-waiting thread, if one exists, from the
     * wait queue for this condition to the wait queue for the
     * owning lock.
     *
     * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
     *         returns {@code false}
     */
    public final void signal() {
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        // firstWaiter 肯定是最早开始等待的
        Node first = firstWaiter;
        // 如果不为空就唤醒
        if (first != null)
            doSignal(first);
    }
    /**
     * Removes and transfers nodes until hit non-cancelled one or
     * null. Split out from signal in part to encourage compilers
     * to inline the case of no waiters.
     * @param first (non-null) the first node on condition queue
     */
    private void doSignal(Node first) {
        do {
            // 因为要去唤醒 first 节点了，firstWaiter 需要再从后面找一个
            // 并且判断是否为空，如果是空的话就直接可以把 lastWaiter 设置成空了
            if ( (firstWaiter = first.nextWaiter) == null)
                lastWaiter = null;
            //  first 不需要继续保存后面的 waiter 了，因为 firstWaiter 已经是 first 的后置节点了
            first.nextWaiter = null;
        // 如果 first 节点转移不成功，并且 firstWaiter 节点不为空，则继续进入循环
        } while (!transferForSignal(first) &&
                 (first = firstWaiter) != null);
    }
    /**
     * Transfers a node from a condition queue onto sync queue.
     * Returns true if successful.
     * @param node the node
     * @return true if successfully transferred (else the node was
     * cancelled before signal)
     */
    final boolean transferForSignal(Node node) {
        /*
         * If cannot change waitStatus, the node has been cancelled.
         */
        // 如果状态已经不是 CONDITION 就不会设置成功，返回 false
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        /*
         * Splice onto queue and try to set waitStatus of predecessor to
         * indicate that thread is (probably) waiting. If cancelled or
         * attempt to set waitStatus fails, wake up to resync (in which
         * case the waitStatus can be transiently and harmlessly wrong).
         */
        // 调用跟aqs 第一篇中一样的 enq 方法进入阻塞队列，返回入队后的前一节点
        Node p = enq(node);
        int ws = p.waitStatus;
        // 将前置节点状态设置成SIGNAL，表示后面有节点在等了
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        // 返回 true，上一个方法的循环就退出了
        return true;
    }

这里其实就是把 condition 等待队列的第一个未取消的节点入队到阻塞队列去争锁

## 附录

`synchronized` 版的 BoundedBuffer

    /*
      File: BoundedBuffer.java

      Originally written by Doug Lea and released into the public domain.
      This may be used for any purposes whatsoever without acknowledgment.
      Thanks for the assistance and support of Sun Microsystems Labs,
      and everyone contributing, testing, and using this code.

      History:
      Date       Who                What
      11Jun1998  dl               Create public version
      17Jul1998  dl               Simplified by eliminating wait counts
      25aug1998  dl               added peek
       5May1999  dl               replace % with conditional (slightly faster)
    */

    package EDU.oswego.cs.dl.util.concurrent;

    /**
     * Efficient array-based bounded buffer class.
     * Adapted from CPJ, chapter 8, which describes design.
     * <p>[<a href="http://gee.cs.oswego.edu/dl/classes/EDU/oswego/cs/dl/util/concurrent/intro.html"> Introduction to this package. </a>] <p>
     **/

    public class BoundedBuffer implements BoundedChannel {

      protected final Object[]  array_;      // the elements

      protected int takePtr_ = 0;            // circular indices
      protected int putPtr_ = 0;

      protected int usedSlots_ = 0;          // length
      protected int emptySlots_;             // capacity - length

      /**
       * Helper monitor to handle puts.
       **/
      protected final Object putMonitor_ = new Object();

      /**
       * Create a BoundedBuffer with the given capacity.
       * @exception IllegalArgumentException if capacity less or equal to zero
       **/
      public BoundedBuffer(int capacity) throws IllegalArgumentException {
        if (capacity <= 0) throw new IllegalArgumentException();
        array_ = new Object[capacity];
        emptySlots_ = capacity;
      }

      /**
       * Create a buffer with the current default capacity
       **/

      public BoundedBuffer() {
        this(DefaultChannelCapacity.get());
      }

      /**
       * Return the number of elements in the buffer.
       * This is only a snapshot value, that may change
       * immediately after returning.
       **/
      public synchronized int size() { return usedSlots_; }

      public int capacity() { return array_.length; }

      protected void incEmptySlots() {
        synchronized(putMonitor_) {
          ++emptySlots_;
          putMonitor_.notify();
        }
      }

      protected synchronized void incUsedSlots() {
        ++usedSlots_;
        notify();
      }

      protected final void insert(Object x) { // mechanics of put
        --emptySlots_;
        array_[putPtr_] = x;
        if (++putPtr_ >= array_.length) putPtr_ = 0;
      }

      protected final Object extract() { // mechanics of take
        --usedSlots_;
        Object old = array_[takePtr_];
        array_[takePtr_] = null;
        if (++takePtr_ >= array_.length) takePtr_ = 0;
        return old;
      }

      public Object peek() {
        synchronized(this) {
          if (usedSlots_ > 0)
            return array_[takePtr_];
          else
            return null;
        }
      }

      public void put(Object x) throws InterruptedException {
        if (x == null) throw new IllegalArgumentException();
        if (Thread.interrupted()) throw new InterruptedException();

        synchronized(putMonitor_) {
          while (emptySlots_ <= 0) {
    	try { putMonitor_.wait(); }
            catch (InterruptedException ex) {
              putMonitor_.notify();
              throw ex;
            }
          }
          insert(x);
        }
        incUsedSlots();
      }

      public boolean offer(Object x, long msecs) throws InterruptedException {
        if (x == null) throw new IllegalArgumentException();
        if (Thread.interrupted()) throw new InterruptedException();

        synchronized(putMonitor_) {
          long start = (msecs <= 0)? 0 : System.currentTimeMillis();
          long waitTime = msecs;
          while (emptySlots_ <= 0) {
            if (waitTime <= 0) return false;
    	try { putMonitor_.wait(waitTime); }
            catch (InterruptedException ex) {
              putMonitor_.notify();
              throw ex;
            }
            waitTime = msecs - (System.currentTimeMillis() - start);
          }
          insert(x);
        }
        incUsedSlots();
        return true;
      }

      public  Object take() throws InterruptedException {
        if (Thread.interrupted()) throw new InterruptedException();
        Object old = null;
        synchronized(this) {
          while (usedSlots_ <= 0) {
            try { wait(); }
            catch (InterruptedException ex) {
              notify();
              throw ex;
            }
          }
          old = extract();
        }
        incEmptySlots();
        return old;
      }

      public  Object poll(long msecs) throws InterruptedException {
        if (Thread.interrupted()) throw new InterruptedException();
        Object old = null;
        synchronized(this) {
          long start = (msecs <= 0)? 0 : System.currentTimeMillis();
          long waitTime = msecs;

          while (usedSlots_ <= 0) {
            if (waitTime <= 0) return null;
            try { wait(waitTime); }
            catch (InterruptedException ex) {
              notify();
              throw ex;
            }
            waitTime = msecs - (System.currentTimeMillis() - start);

          }
          old = extract();
        }
        incEmptySlots();
        return old;
      }

    }

## AQS篇一

发表于 2021-02-14 分类于 Java ， 并发 阅读次数： Disqus：

很多东西都是时看时新，而且时间长了也会忘，所以再来复习下，也会有一些新的角度看法这次来聊下AQS的内容，主要是这几个点，

## 第一个线程

第一个线程抢到锁了，此时state跟阻塞队列是怎么样的，其实这里是之前没理解对的地方

    /**
             * Fair version of tryAcquire.  Don't grant access unless
             * recursive call or no waiters or is first.
             */
            protected final boolean tryAcquire(int acquires) {
                final Thread current = Thread.currentThread();
                int c = getState();
                // 这里如果state还是0说明锁还空着
                if (c == 0) {
                    // 因为是公平锁版本的，先去看下是否阻塞队列里有排着队的
                    if (!hasQueuedPredecessors() &&
                        compareAndSetState(0, acquires)) {
                        // 没有排队的，并且state使用cas设置成功的就标记当前占有锁的线程是我
                        setExclusiveOwnerThread(current);
                        // 然后其实就返回了，包括阻塞队列的head和tail节点和waitStatus都没有设置
                        return true;
                    }
                }
                else if (current == getExclusiveOwnerThread()) {
                    int nextc = c + acquires;
                    if (nextc < 0)
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                // 这里就是第二个线程会返回false
                return false;
            }
        }

## 第二个线程

当第二个线程进来的时候应该是怎么样，结合代码来看

    /**
         * Acquires in exclusive mode, ignoring interrupts.  Implemented
         * by invoking at least once {@link #tryAcquire},
         * returning on success.  Otherwise the thread is queued, possibly
         * repeatedly blocking and unblocking, invoking {@link
         * #tryAcquire} until success.  This method can be used
         * to implement method {@link Lock#lock}.
         *
         * @param arg the acquire argument.  This value is conveyed to
         *        {@link #tryAcquire} but is otherwise uninterpreted and
         *        can represent anything you like.
         */
        public final void acquire(int arg) {
            // 前面第一种情况是tryAcquire直接成功了，这个if判断第一个条件就是false，就不往下执行了
            // 如果是第二个线程，第一个条件获取锁不成功，条件判断!tryAcquire(arg) == true，就会走
            // acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }

然后来看下addWaiter的逻辑

    /**
         * Creates and enqueues node for current thread and given mode.
         *
         * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
         * @return the new node
         */
        private Node addWaiter(Node mode) {
            // 这里是包装成一个node
            Node node = new Node(Thread.currentThread(), mode);
            // Try the fast path of enq; backup to full enq on failure
            // 最快的方式就是把当前线程的节点放在阻塞队列的最后
            Node pred = tail;
            // 只有当tail，也就是pred不为空的时候可以直接接上
            if (pred != null) {
                node.prev = pred;
                // 如果这里cas成功了，就直接接上返回了
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
            // 不然就会继续走到这里
            enq(node);
            return node;
        }

然后就是enq的逻辑了

    /**
         * Inserts node into queue, initializing if necessary. See picture above.
         * @param node the node to insert
         * @return node's predecessor
         */
        private Node enq(final Node node) {
            for (;;) {
                // 如果状态没变化的话，tail这时还是null的
                Node t = tail;
                if (t == null) { // Must initialize
                    // 这里就会初始化头结点，就是个空节点
                    if (compareAndSetHead(new Node()))
                        // tail也赋值成head
                        tail = head;
                } else {
                    // 这里就设置tail了
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }

所以从这里可以看出来，其实head头结点不是个真实的带有线程的节点，并且不是在第一个线程进来的时候设置的

## 解锁

通过代码来看下

    /**
         * Attempts to release this lock.
         *
         * <p>If the current thread is the holder of this lock then the hold
         * count is decremented.  If the hold count is now zero then the lock
         * is released.  If the current thread is not the holder of this
         * lock then {@link IllegalMonitorStateException} is thrown.
         *
         * @throws IllegalMonitorStateException if the current thread does not
         *         hold this lock
         */
        public void unlock() {
            // 释放锁
            sync.release(1);
        }
    /**
         * Releases in exclusive mode.  Implemented by unblocking one or
         * more threads if {@link #tryRelease} returns true.
         * This method can be used to implement method {@link Lock#unlock}.
         *
         * @param arg the release argument.  This value is conveyed to
         *        {@link #tryRelease} but is otherwise uninterpreted and
         *        can represent anything you like.
         * @return the value returned from {@link #tryRelease}
         */
        public final boolean release(int arg) {
            // 尝试去释放
            if (tryRelease(arg)) {
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }
    protected final boolean tryRelease(int releases) {
                int c = getState() - releases;
                if (Thread.currentThread() != getExclusiveOwnerThread())
                    throw new IllegalMonitorStateException();
                boolean free = false;
        		// 判断是否完全释放锁，因为可重入
                if (c == 0) {
                    free = true;
                    setExclusiveOwnerThread(null);
                }
                setState(c);
                return free;
            }
    // 这段代码和上面的一致，只是为了顺序性，又拷下来看下

    public final boolean release(int arg) {
            // 尝试去释放，如果是完全释放，返回的就是true，否则是false
            if (tryRelease(arg)) {
                Node h = head;
                // 这里判断头结点是否为空以及waitStatus的状态，前面说了head节点其实是
                // 在第二个线程进来的时候初始化的，如果是空的话说明没后续节点，并且waitStatus
                // 也表示了后续的等待状态
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }

    /**
         * Wakes up node's successor, if one exists.
         *
         * @param node the node
         */
    // 唤醒后继节点
        private void unparkSuccessor(Node node) {
            /*
             * If status is negative (i.e., possibly needing signal) try
             * to clear in anticipation of signalling.  It is OK if this
             * fails or if status is changed by waiting thread.
             */
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);

            /*
             * Thread to unpark is held in successor, which is normally
             * just the next node.  But if cancelled or apparently null,
             * traverse backwards from tail to find the actual
             * non-cancelled successor.
             */
            Node s = node.next;
            // 如果后继节点是空或者当前节点取消等待了
            if (s == null || s.waitStatus > 0) {
                s = null;
                // 从后往前找，找到非取消的节点，注意这里不是找到就退出，而是一直找到头
                // 所以不必担心中间有取消的
                for (Node t = tail; t != null && t != node; t = t.prev)
                    if (t.waitStatus <= 0)
                        s = t;
            }
            if (s != null)
                // 将其唤醒
                LockSupport.unpark(s.thread);
        }

## 关于读书打卡与分享

发表于 2021-02-07 分类于 生活 ， 读后感 ， 白岩松 ， 幸福了吗 阅读次数： Disqus：

最近群里大佬发起了一个读书打卡活动，需要每天读一会书，在群里打卡分享感悟，争取一个月能读完一本书，说实话一天十分钟的读书时间倒是问题不大，不过每天都要打卡，而且一个月要读完一本书，其实难度还是有点大的，不过也想试试看。
之前某某老大给自己立了个 flag，说要读一百本书，这对我来说挺难实现的，一则我也不喜欢书只读一小半，二则感觉对于喜欢看的内容范围还是比较有限制，可能也算是比较矫情，不爱追热门的各类东西，因为往往会有一些跟大众不一致的观点看法，显得格格不入。所以还是这个打卡活动可能会比较适合我，书是人类进步的阶梯。
到现在是打卡了三天了，读的主要是白岩松的《幸福了吗》，对于白岩松，我们这一代人是比较熟悉，并且整体印象比较不错的一个央视主持人，从《焦点访谈》开始，到疫情期间的各类一线节目，可能对我来说是个三观比较正，敢于说一些真话的主持人，这中间其实是有个空档期，没怎么看电视，也不太关注了，只是在疫情期间的节目，还是一如既往地给人一种可靠的感觉，正好有一次偶然微信读书推荐了白岩松的这本书，就看了一部分，正好这次继续往下看，因为相对来讲不会很晦涩，并且从这位知名央视主持人的角度分享他的过往和想法看法，还是比较有意思的。
从对汶川地震，08 年奥运等往事的回忆和一些细节的呈现，也让我了解比较多当时所不知道的，特别是汶川地震，那时的我还在读高中，真的是看着电视，作为“猛男”都忍不住泪目了，共和国之殇，多难兴邦，但是这对于当事人来说，都是一场醒不过来的噩梦。
然后是对于足球的热爱，其实光这个就能掰扯很多，因为我不爱足球，只爱篮球，其中原因有的没的也挺多可以说的，但是看了他的书，才能比较深入的了解一个足球迷，对足球，对中国足球，对世界杯，对阿根廷的感情。
接下去还是想能继续坚持下去，加油！

## 聊聊 redis 缓存的应用问题

发表于 2021-01-31 分类于 Redis ， 应用 ， 缓存 ， 缓存 ， 穿透 ， 击穿 ， 雪崩 阅读次数： Disqus：

前面写过一系列的 redis 源码分析的，但是实际上很多的问题还是需要结合实际的使用，然后其实就避不开缓存使用的三个著名问题，穿透，击穿和雪崩，这三个概念也是有着千丝万缕的关系，

### 缓存穿透

缓存穿透是指当数据库中本身就不存在这个数据的时候，使用一般的缓存策略时访问不到缓存后就访问数据库，但是因为数据库也没数据，所以如果不做任何策略优化的话，这类数据就每次都会访问一次数据库，对数据库压力也会比较大。

### 缓存击穿

缓存击穿跟穿透比较类似的，都是访问缓存不在，然后去访问数据库，与穿透不一样的是击穿是在数据库中存在数据，但是可能由于第一次访问，或者缓存过期了，需要访问到数据库，这对于访问量小的情况其实算是个正常情况，但是随着请求量变高就会引发一些性能隐患。

### 缓存雪崩

缓存雪崩就是击穿的大规模集群效应，当大量的缓存过期失效的时候，这些请求都是直接访问到数据库了，会对数据库造成很大的压力。

对于以上三种场景也有一些比较常见的解决方案，但也不能说是万无一失的，需要随着业务去寻找合适的方案

### 解决缓存穿透

对于数据库中就没这个数据的时候，一种是可以对这个 key 设置下空值，即以一个特定的表示是数据库不存在的，这种情况需要合理地调整过期时间，当这个 key 在数据库中有数据了的话，也需要有策略去更新这个值，并且如果这类 key 非常多，这个方法就会不太合适，就可以使用第二种方法，就是布隆过滤器，bloom filter，前置一个布隆过滤器，当这个 key 在数据库不存在的话，先用布隆过滤器挡一道，如果不在的话就直接返回了，当然布隆过滤器不是绝对的准确的

### 解决缓存击穿

当一个 key 的缓存过期了，如果大量请求过来访问这个 key，请求都会落在数据库里，这个时候就可以使用一些类似于互斥锁的方式去让一个线程去访问数据库，更新缓存，但是这里其实也有个问题，就是如果是热点 key 其实这种方式也比较危险，万一更新失败，或者更新操作的时候耗时比较久，就会有一大堆请求卡在那，这种情况可能需要有一些异步提前刷新缓存，可以结合具体场景选择方式

### 解决缓存雪崩

雪崩的情况是指大批量的 key 都一起过期了，击穿的放大版，大批量的请求都打到数据库上了，一方面有可能直接缓存不可用了，就需要用集群化高可用的缓存服务，然后对于实际使用中也可以使用本地缓存结合 redis 缓存，去提高可用性，再配合一些限流措施，然后就是缓存使用过程总的过期时间最好能加一些随机值，防止在同一时间过期而导致雪崩，结合互斥锁防止大量请求打到数据库。

## Leetcode 124 二叉树中的最大路径和(Binary Tree Maximum Path Sum) 题解分析

发表于 2021-01-24 分类于 Java ， leetcode ， Binary Tree ， java ， Binary Tree 阅读次数： Disqus：

### 题目介绍

A path in a binary tree is a sequence of nodes where each pair of adjacent nodes in the sequence has an edge connecting them. A node can only appear in the sequence at most once. Note that the path does not need to pass through the root.

The path sum of a path is the sum of the node’s values in the path.

Given the `root` of a binary tree, return the maximum path sum of any path.

路径 被定义为一条从树中任意节点出发，沿父节点-子节点连接，达到任意节点的序列。该路径 至少包含一个 节点，且不一定经过根节点。

路径和 是路径中各节点值的总和。

给你一个二叉树的根节点 `root` ，返回其 最大路径和

### 简要分析

其实这个题目会被误解成比较简单，左子树最大的，或者右子树最大的，或者两边加一下，仔细想想都不对，其实有可能是产生于左子树中，或者右子树中，这两个都是指跟左子树根还有右子树根没关系的，这么说感觉不太容易理解，画个图
可以看到图里，其实最长路径和是左边这个子树组成的，跟根节点还有右子树完全没关系，然后再想一种情况，如果是整棵树就是图中的左子树，那么这个最长路径和就是左子树加右子树加根节点了，所以不是我一开始想得那么简单，在代码实现中也需要一些技巧

### 代码

    int ansNew = Integer.MIN_VALUE;
    public int maxPathSum(TreeNode root) {
            maxSumNew(root);
            return ansNew;
        }

    public int maxSumNew(TreeNode root) {
        if (root == null) {
            return 0;
        }
        // 这里是个简单的递归，就是去递归左右子树，但是这里其实有个概念，当这样处理时，其实相当于把子树的内部的最大路径和已经算出来了
        int left = maxSumNew(root.left);
        int right = maxSumNew(root.right);
        // 这里前面我有点没想明白，但是看到 ansNew 的比较，其实相当于，返回的是三种情况里的最大值，一个是左子树+根，一个是右子树+根，一个是单独根节点，
        // 这样这个递归的返回才会有意义，不然像原来的方法，它可能是跳着的，但是这种情况其实是借助于 ansNew 这个全局的最大值，因为原来我觉得要比较的是
        // left, right, left + root , right + root, root, left + right + root 这些的最大值，这里是分成了两个阶段，left 跟 right 的最大值已经在上面的
        // 调用过程中赋值给 ansNew 了
        int currentSum = Math.max(Math.max(root.val + left , root.val + right), root.val);
        // 这边返回的是 currentSum，然后再用它跟 left + right + root 进行对比，然后再去更新 ans
        // PS: 有个小点也是这边的破局点，就是这个 ansNew
        int res = Math.max(left + right + root.val, currentSum);
        ans = Math.max(res, ans);
        return currentSum;
    }

这里非常重要的就是 ansNew 是最后的一个结果，而对于 maxSumNew 这个函数的返回值其实是需要包含了一个连续结果，因为要返回继续去算路径和，所以返回的是 currentSum，最终结果是 ansNew

### 结果图

难得有个 100%，贴个图哈哈