• 介绍
• 源码分析
  • Executor
  • ExecutorService
  • AbstractExecutorService#newTaskFor
  • AbstractExecutorService#submit
  • AbstractExecutorService#doInvokeAny()
  • AbstractExecutorService#invokeAny()
  • AbstractExecutorService#invokeAll()
  • ThreadPoolExecutor
  • ThreadPoolExecutor 构造函数
  • Executors#defaultThreadFactory()
  • Executors#DefaultThreadFactory
  • ThreadPoolExecutor#execute()
  • ThreadPoolExecutor#addIfUnderCorePoolSize()
  • ThreadPoolExecutor#addThread()
  • ThreadPoolExecutor#Worker
  • ThreadPoolExecutor#beforeExecute()
  • ThreadPoolExecutor#afterExecute()
  • ThreadPoolExecutor#getTask()
  • ThreadPoolExecutor#workerCanExit()
  • ThreadPoolExecutor#interruptIdleWorkers()
  • ThreadPoolExecutor#workerDone()
  • ThreadPoolExecutor#tryTerminate()
  • ThreadPoolExecutor#ensureQueuedTaskHandled()
  • ThreadPoolExecutor#addIfUnderMaximumPoolSize()
  • ThreadPoolExecutor#shutdown()
  • ThreadPoolExecutor#shutdownNow()
  • ThreadPoolExecutor#drainQueue()
  • ThreadPoolExecutor#awaitTermination()
  • ThreadPoolExecutor#finalize()
  • ThreadPoolExecutor#prestartCoreThread()
  • ThreadPoolExecutor#prestartAllCoreThreads()
  • ThreadPoolExecutor#setCorePoolSize()
  • ThreadPoolExecutor#setMaximumPoolSize()
  • 实现RejectedExecutionHandler接口
    • ThreadPoolExecutor#AbortPolicy
    • ThreadPoolExecutor#CallerRunsPolicy
    • ThreadPoolExecutor#DiscardPolicy
    • ThreadPoolExecutor#DiscardOldestPolicy

介绍

ThreadPoolExecutor是JUC包中提供的线程池，使用ThreadPoolExecutor的好处一方面是能重用线程资源，避免重复创建线程带来的开销；另一方面是ThreadPoolExecutor提供了内部资源(线程、任务)的管理功能，方便我们监控线程池工作状态。

源码分析

ThreadPoolExecutor类结构图

Executor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

public interface Executor {

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the <tt>Executor</tt> implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution.
     * @throws NullPointerException if command is null
     */
    // 1
    void execute(Runnable command);
}

标注代码分析

1. 根据注释很好理解。command可以在新线程里执行，可以在线程池里执行，可以被调用线程执行等等。

• 介绍
• 源码分析
  • SynchronousQueue#Transferer
  • SynchronousQueue#TransferStack 伪栈实现
  • SynchronousQueue#TransferStack#transfer()
  • SynchronousQueue#TransferStack#snode()
  • SynchronousQueue#TransferStack#awaitFulfill()
  • SynchronousQueue#TransferStack#shouldSpin()
  • SynchronousQueue自选参数
  • SynchronousQueue#TransferStack#clean()
  • SynchronousQueue#TransferQueue 伪队列
  • SynchronousQueue#TransferQueue#transfer()
  • SynchronousQueue#TransferQueue#advanceHead()
  • SynchronousQueue#TransferQueue#advanceTail()
  • SynchronousQueue#TransferQueue#awaitFulfill()
  • SynchronousQueue#TransferQueue#clean()
  • SynchronousQueue
  • SynchronousQueue 序列化
• 总结

介绍

• SynchronousQueue是一种特殊的阻塞队列，它本身没有容量，只有当一个线程从队列取数据的同时，另一个线程才能放一个数据到队列中，反之亦然。存取过程相当于一个线程把数据(安全的)交给另一个线程的过程。
• SynchronousQueue也支持公平和非公平模式。

• 例
• 源码分析
  • ThreadLocal#Entry
  • ThreadLocal#nextHashCode
  • ThreadLocal#HASH_INCREMENT
  • ThreadLocal#threadLocalHashCode
  • Thread#threadLocals
  • ThreadLocal#get()
  • ThreadLocal#getMap()
  • ThreadLocal#ThreadLocalMap#getEntry()
  • ThreadLocal#set()
  • ThreadLocal#createMap()
  • ThreadLocal#ThreadLocalMap#set()
  • ThreadLocal#ThreadLocalMap#replaceStaleEntry()
  • ThreadLocal#ThreadLocalMap#expungeStaleEntry()
  • ThreadLocal#ThreadLocalMap#cleanSomeSlots()
• 程序实例
  • 同一个线程多个值
  • 多个线程
• 总结

例

Hibernate中典型的ThreadLocal的代码。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

private static final ThreadLocal threadSession = new ThreadLocal();  
  
public static Session getSession() throws InfrastructureException {  
    Session s = (Session) threadSession.get();  
    try {  
        if (s == null) {  
            s = getSessionFactory().openSession();  
            threadSession.set(s);  
        }  
    } catch (HibernateException ex) {  
        throw new InfrastructureException(ex);  
    }  
    return s;  
}

可以看到在getSession()方法中，首先判断当前线程中有没有放进去session，如果还没有，那么通过sessionFactory().openSession()来创建一个session，再将session set到线程中，实际是放到当前线程的ThreadLocalMap这个map中。这时，对于这个session的唯一引用就是当前线程中的那个ThreadLocalMap，而threadSession作为这个值的key，要取得这个session可以通过threadSession.get()来得到，里面执行的操作实际是先取得当前线程中的ThreadLocalMap，然后将threadSession作为key将对应的值取出。这个session相当于线程的私有变量，而不是public的。
ThreadLocal是在每个线程中有一个map，而将ThreadLocal实例作为key，这样每个map中的项数很少，而且当线程销毁时相应的东西也一起销毁了。 这样设计也防止内存泄漏，1个线程被销毁，这个线程内的对象都被销毁。

• 介绍
  • 优点
• 源码分析
  • ScheduledExecutorService()
  • ScheduledThreadPoolExecutor#属性
  • ScheduledThreadPoolExecutor()
  • ScheduledThreadPoolExecutor#DelayedWorkQueue()
  • ScheduledThreadPoolExecutor#schedule()
  • ScheduledThreadPoolExecutor#triggerTime()
  • ScheduledThreadPoolExecutor#overflowFree()
  • ScheduledThreadPoolExecutor#decorateTask()
  • ScheduledThreadPoolExecutor#ScheduledFutureTask
  • RunnableScheduledFuture
  • ScheduledThreadPoolExecutor#ScheduledFutureTask#getDelay()
  • ScheduledThreadPoolExecutor#ScheduledFutureTask#compareTo()
  • ScheduledThreadPoolExecutor#getQueue()
  • ScheduledThreadPoolExecutor#delayedExecute()
  • ScheduledThreadPoolExecutor#ScheduledFutureTask#delayedExecute()
  • ScheduledThreadPoolExecutor#ScheduledFutureTask#run()
  • ScheduledThreadPoolExecutor#ScheduledFutureTask#runPeriodic()
  • ScheduledThreadPoolExecutor#scheduleAtFixedRate()
  • ScheduledThreadPoolExecutor#scheduleWithFixedDelay()
  • ScheduledThreadPoolExecutor#setContinueExistingPeriodicTasksAfterShutdownPolicy()
  • ScheduledThreadPoolExecutor#setExecuteExistingDelayedTasksAfterShutdownPolicy()
  • ScheduledThreadPoolExecutor#shutdown()
  • ScheduledThreadPoolExecutor#cancelUnwantedTasks()
• 总结

介绍

ScheduledThreadPoolExecutor是一种类似Timer的定时器或者说是调度器。

优点

1. 多线程的定时调度，timer是单线程的，每个timer实例只有一个工作线程。
2. 由于继承自ThreadPoolExecutor，更具有灵活性和伸缩性。
3. 没有timer那种线程泄露问题，timer调度的任务如果异常终止，那么整个timer都会被取消，无法执行其他任务。

• 介绍
  • 写锁(tryAcquire)
  • 读锁(tryAcquireShared)
• 源码分析
  • Sync属性
  • Sync#tryAcquire()
  • Sync#writerShouldBlock()
  • Sync#tryRelease()
  • Sync#HoldCounter
  • Sync#ThreadLocalHoldCounter
  • Sync#tryAcquireShared
  • Sync#fullTryAcquireShared
  • Sync#tryReleaseShared
  • Sync#tryWriteLock
  • Sync#tryReadLock
  • ReentrantReadWriteLock#FairSync
  • ReentrantReadWriteLock#NonfairSync
  • ReentrantReadWriteLock#ReadLock#lock()
  • ReentrantReadWriteLock#ReadLock#tryLock()
  • ReentrantReadWriteLock#ReadLock#newCondition()
  • ReentrantReadWriteLock#WriteLock#lock()
  • ReentrantReadWriteLock#WriteLock#tryLock()
• 锁降级

介绍

ReentrantReadWriteLock支持公平/非公平锁。写锁可以降级为读锁，但读锁不能升级为写锁。

写锁(tryAcquire)

(读锁&&写锁)或者(无读锁&&无写锁)，根据公平锁判断queue empty或者queue head，再竞争锁，非公平锁直接竞争锁。

读锁(tryAcquireShared)

读锁会阻塞写锁。也会根据公平锁判断queue empty或者queue head，再竞争锁，非公平锁队列头有写请求，那么当前读请求阻塞，返回false，CAS直接竞争锁。如果是queue head，返回true，通过fullTryAcquireShared()中的CAS竞争锁。

源码分析

ReentrantReadWriteLock实现ReadWriteLock接口。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

public interface ReadWriteLock {
    /**
     * Returns the lock used for reading.
     *
     * @return the lock used for reading.
     */
    Lock readLock();

    /**
     * Returns the lock used for writing.
     *
     * @return the lock used for writing.
     */
    Lock writeLock();
}

读锁和写锁2个方法。
ReentrantReadWriteLock#Sync基于AbstractQueuedSynchronizer实现。

• 介绍
  • 优势
  • 缺点
• 源码分析
  • Lock
  • ReentrantLock#Sync
  • ReentrantLock#Sync#nonfairTryAcquire()
  • ReentrantLock#Sync#tryRelease()
  • ReentrantLock#NonfairSync
  • ReentrantLock#FairSync
  • ReentrantLock构造方法
  • ReentrantLock#getHoldCount()

介绍

ReentrantLock和Synchronized一样都是可重入的。
ReentrantLock与Synchronized相比较而言。

优势

支持公平/非公平锁、支持可中断的锁、支持非阻塞的tryLock(可超时)、支持锁条件、可跨代码块使用(一个地方加锁，另一个地方解锁)，总之比Synchronized更加灵活。

缺点

锁需要显示解锁、无法充分享用JVM内部性能提升带来的好处等等。

源码分析

ReentrantLock实现Lock接口。

Lock

1
2
3
4
5
6
7
8
9
10
11
12
13

public interface Lock {
    // 1
    void lock();
    // 2
    void lockInterruptibly() throws InterruptedException;
    // 3
    boolean tryLock();
    // 4
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
  
    void unlock();
    Condition newCondition();
}

标注代码分析

1. 获取锁，如果锁无法获取，当前线程被阻塞，直到锁可以获取并获取成功为止。
2. 在当前线程没有被中断的情况下获取锁。如果锁无法获取，当前线程被阻塞，阻塞获取锁或者中断。此处有抛出InterruptedException。
3. 尝试获取锁。
4. 有超时时间和抛出InterruptedException。在指定时间内获取锁或者线程中断。

• 详解
  • HashMap#indexFor()
  • HashMap#hash()
  • Redis2.4
  • HashMap
  • redis
  • 例
  • 反编译
• 总结

详解

HashMap#indexFor()

1
2
3
4
5
6
7

/**
 * Returns index for hash code h.
 */
static int indexFor(int h, int length) {
    // 1
    return h & (length-1);
}

HashMap#hash()

1
2
3
4
5
6
7
8
9
10
11
12
13
14

/**
 * Applies a supplemental hash function to a given hashCode, which
 * defends against poor quality hash functions.  This is critical
 * because HashMap uses power-of-two length hash tables, that
 * otherwise encounter collisions for hashCodes that do not differ
 * in lower bits. Note: Null keys always map to hash 0, thus index 0.
 */
static int hash(int h) {
    // This function ensures that hashCodes that differ only by
    // constant multiples at each bit position have a bounded
    // number of collisions (approximately 8 at default load factor).
    h ^= (h >>> 20) ^ (h >>> 12);
    return h ^ (h >>> 7) ^ (h >>> 4);
}

• 介绍
• 源码分析
  • LinkedBlockingDeque
  • LinkedBlockingDeque#linkFirst()
  • LinkedBlockingDeque#linkLast()
  • LinkedBlockingDeque#unlinkFirst()
  • LinkedBlockingDeque#unlinkLast()
  • LinkedBlockingDeque#unlink()
  • LinkedBlockingDeque#putFirst()
  • LinkedBlockingDeque#putLast()
  • LinkedBlockingDeque#takeFirst()
  • LinkedBlockingDeque#takeLast()
  • LinkedBlockingDeque#iterator()
  • LinkedBlockingDeque#Itr

介绍

LinkedBlockingDeque是一种基于双向链表实现的有界的(可选的，不指定默认int最大值)阻塞双端队列。
双端队列一般适用于工作密取模式，即每个消费者都拥有自己的双端队列，如果某个消费者完成了自己队列的全部任务，可以到其他消费者双端队列尾部秘密获取任务来处理。

• 介绍
• 源码分析
  • LinkedBlockingQueue
  • LinkedBlockingQueue#put()
  • LinkedBlockingQueue#enqueue()
  • LinkedBlockingQueue#signalNotEmpty()
  • LinkedBlockingQueue#take()
  • LinkedBlockingQueue#dequeue()
  • LinkedBlockingQueue#signalNotFull()
  • LinkedBlockingQueue#remove()
  • LinkedBlockingQueue#fullyLock()
  • LinkedBlockingQueue#fullyUnlock()

介绍

LinkedBlockingQueue是一种基于单向链表实现的有界的(可选的，不指定默认int最大值)阻塞队列。队列中的元素遵循先入先出(FIFO)的规则。新元素插入到队列的尾部，从队列头部取出元素。(在并发程序中，基于链表实现的队列和基于数组实现的队列相比，往往具有更高的吞吐量，但性能稍差一些)

• 功能简介
• 源码分析
  • RunnableFuture接口
  • Runnable接口
  • Future接口
  • FutureTask#Sync
  • Callable接口
  • FutureTask#innerRun()
  • FutureTask#Sync#innerSet()
  • AbstractQueuedSynchronizer#releaseShared()
  • AbstractQueuedSynchronizer#tryReleaseShared()
  • FutureTask#Sync#tryReleaseShared()
  • FutureTask#Sync#innerSetException
  • FutureTask#Sync#innerRunAndReset()
  • FutureTask#Sync#innerGet()
  • FutureTask#Sync#innerGet(long nanosTimeout)
  • FutureTask#get()
  • AbstractQueuedSynchronizer#acquireSharedInterruptibly()
  • AbstractQueuedSynchronizer#tryAcquireShared()
  • FutureTask#Sync#tryAcquireShared()
  • FutureTask#Sync#innerIsDone()
  • FutureTask#Sync#ranOrCancelled()
  • FutureTask#Sync#innerCancel()
  • FutureTask#cancel()
  • FutureTask
  • FutureTask构造方法
  • Executors#callable()
  • Executors#RunnableAdapter
• 3 总结

功能简介

• FutureTask是一种异步任务(或异步计算)，举个栗子，主线程的逻辑中需要使用某个值，但这个值需要复杂的运算得来，那么主线程可以提前建立一个异步任务来计算这个值(在其他的线程中计算)，然后去做其他事情，当需要这个值的时候再通过刚才建立的异步任务来获取这个值，有点并行的意思，这样可以缩短整个主线程逻辑的执行时间。
• FutureTask是基于AQS来构建的，使用共享模式，使用AQS的状态来表示异步任务的运行状态。