多线程进阶篇

1、JUC原子类

• 基本类型: AtomicInteger, AtomicLong, AtomicBoolean ;
• 数组类型: AtomicIntegerArray, AtomicLongArray, AtomicReferenceArray ;
• 引用类型: AtomicReference, AtomicStampedRerence, AtomicMarkableReference ;
• 对象的属性修改类型: AtomicIntegerFieldUpdater, AtomicLongFieldUpdater, AtomicReferenceFieldUpdater 。

2、JUC锁

2.1、互斥锁

ReentrantLock是Lock的唯一实现类，而Condition:依赖ReentrantLock,一个ReentrantLock对象可以有多个Condition

2.2、读写锁：ReentrantReadWriteLock是ReadWriteLock的实现类

2.3、 公平锁和非公平锁

公平锁：多个线程等待一个锁，等待最久的线程先获取锁。

ReentrantLock和ReentrantReadWriteLock可以设置成公平锁

非公平锁：跟公平锁相反，无法保证等待最久的线程先获取锁。

synchronized是非公平锁，ReentrantLock和ReentrantReadWriteLock默认是非公平锁

以ReentrantLock为例，ReentrantLock与Sync是组合关系，Sync是AQS的子类，Sync有两个子类FairSync(公平锁)和NonFairSync(非公平锁)。
ReentrantLock是公平锁还是非公平锁，就取决于sync对象是"FairSync的实例"还是"NonFairSync的实例"。

2.4、CountDownLatch

要等待其他任务执行完成，才会执行。

例如有一个任务A，它要等待其他4个任务执行完毕之后才能执行。

构造器：

CountDownLatch(int count);  //参数count为计数值

常用方法：

await();   //调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行
await(long timeout, TimeUnit unit);  //和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行
countDown() { };  //将count值减1

2.5、CyclicBarrier

通过它可以实现让一组线程等待至某个状态之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后，CyclicBarrier可以被重用。

构造器：

CyclicBarrier(int parties, Runnable barrierAction)
CyclicBarrier(int parties) 

参数parties指让多少个线程或者任务等待至barrier状态；参数barrierAction为当这些线程都达到barrier状态时会执行的内容。

int await();
int await(long timeout, TimeUnit unit);

不带参await()比较常用，用来挂起当前线程，直至所有线程都到达barrier状态再同时执行后续任务；

带参await()是让这些线程等待至一定的时间，如果还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务。

2.6、Semaphore

Semaphore可以控同时访问的线程个数，通过 acquire() 获取一个许可，如果没有就等待，而 release() 释放一个许可。

Semaphore其实和锁有点类似，它一般用于控制对某组资源的访问权限。

构造器

Semaphore(int permits) //参数permits表示许可数目，即同时可以允许多少线程进行访问，默认设置不公平锁
Semaphore(int permits, boolean fair)  //这个多了一个参数fair表示是否是公平的，即等待时间越久的越先获取许可

常用方法

acquire() //获取一个许可
acquire(int permits) //获取permits个许可
release() //释放一个许可
release(int permits) //释放permits个许可

boolean tryAcquire();//尝试获取一个许可，若获取成功，则立即返回true，若获取失败，则立即返回false
boolean tryAcquire(long timeout, TimeUnit unit);  //尝试获取一个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false
boolean tryAcquire(int permits)//尝试获取permits个许可，若获取成功，则立即返回true，若获取失败，则立即返回false
boolean tryAcquire(int permits, long timeout, TimeUnit unit)//尝试获取permits个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false

availablePermits() //方法得到可用的许可数目

3、JUC集合

4、线程池

image.png

4.1、架构

Executor

是“执行者”接口，用来执行任务。提供execute()接口，来执行已提交的Runnable 任务的对象

ExecutorService

是“执行者服务”接口，提供submit方法将任务提交给执行者
AbstractExecutorService抽象类，实现ExecutorService。提供默认方法

ThreadPoolExecutor线程池

ScheduledExecutorService提供了"延时"和"周期执行"功能的ExecutorService。

ScheduledThreadPoolExecutor

继承ScheduledExecutorService，类似“Timer”，但高并发程序其性能要由于Timer

Executors

是静态工厂类。通过静态工厂方法返回ExecutorService、ScheduledExecutorService、ThreadFactory 和 Callable 等类的对象。

4.2、线程池介绍

ThreadPoolExecutor有四个构造器，事实上都是调用这个构造器

public ThreadPoolExecutor(int corePoolSize,
                         int maximumPoolSize,
                         long keepAliveTime,
                         TimeUnit unit,
                         BlockingQueue<Runnable> workQueue,
                         ThreadFactory threadFactory,
                         RejectedExecutionHandler handler) {

ThreadPoolExecutor的数据结构

image.png

1、workQueue

是BlockingQueue<Runnable>阻塞队列，当线程池数超过容量，线程会进入阻塞队列的等待

2、mainLock

是互斥锁，实现线程池的互斥访问

3、workers

是HashSet<Worker>，是工作线程集合，当执行完一个线程，会从阻塞队列取出一个阻塞线程来运行

4、threadFactory

是ThreadFactory对象。它是一个线程工厂类，"线程池通过ThreadFactory创建线程"。是通过 Executors.defaultThreadFactory()返回的

5、handler 拒绝策略

是RejectedExecutionHandler对象，拒绝策略的处理句柄。

线程池的拒绝策略，是指当任务添加到线程池中被拒绝，而采取的处理措施。

线程池共包括4种拒绝策略

• ThreadPoolExecutor.AbortPolicy -当任务添加到线程池中被拒绝时，它将抛出 RejectedExecutionException 异常。
• ThreadPoolExecutor.CallerRunsPolicy -当任务添加到线程池中被拒绝时，会在线程池当前正在运行的Thread线程池中处理被拒绝的任务。
• ThreadPoolExecutor.DiscardOldestPolicy -当任务添加到线程池中被拒绝时，线程池会放弃等待队列中最旧的未处理任务，然后将被拒绝的任务添加到等待队列中。
• ThreadPoolExecutor.DiscardPolicy -- 当任务添加到线程池中被拒绝时，线程池将丢弃被拒绝的任务。

6、poolSize

poolSize是当前线程池的实际大小，即线程池中任务的数量。

7、allowCoreThreadTimeOut和keepAliveTime

allowCoreThreadTimeOut表示是否允许"线程在空闲状态时，仍然能够存活"；

而keepAliveTime是当线程池处于空闲状态的时候，超过keepAliveTime时间之后，空闲的线程会被终止。

8、corePoolSize

核心池的大小（即线程池中的线程数目大于这个参数时，提交的任务会被放进任务缓存队列）

9、maximumPoolSize

线程池最大线程数，这个参数也是一个非常重要的参数，它表示在线程池中最多能创建多少个线程；

10、largestPoolSize

用来记录线程池中曾经出现过的最大线程数

11、completedTaskCount

用来记录已经执行完毕的任务个数

12、unit

参数keepAliveTime的时间单位

TimeUnit.DAYS;   //天
TimeUnit.HOURS; //小时
TimeUnit.MINUTES;   //分钟
TimeUnit.SECONDS;   //秒
TimeUnit.MILLISECONDS;  //毫秒
TimeUnit.MICROSECONDS;  //微妙
TimeUnit.NANOSECONDS;   //纳秒

13、常用方法

execute()       执行任务
submit()        提交任务，实际上会调用execute()方法
shutdown()      不会立即终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务
shutdownNow()   立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务
setCorePoolSize()        设置核心池大小
setMaximumPoolSize()     设置线程池最大能创建的线程数目大小
prestartCoreThread()     初始化一个核心线程
prestartAllCoreThreads() 初始化所有核心线程

4.2、线程池剖析

4.2.1、线程池状态

线程池状态：Running, SHUTDOWN, STOP, TIDYING, TERMINATED

image.png

如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，此时线程池不能够接受新的任务，它会等待所有任务执行完毕；

如果调用了shutdownNow()方法，则线程池处于STOP状态，此时线程池不能接受新的任务，并且会去尝试终止正在执行的任务；

4.2.2、线程池任务处理策略

• 如果当前线程池中的线程数目小于corePoolSize，则每来一个任务，就会创建一个线程去执行这个任务；
• 如果当前线程池中的线程数目>=corePoolSize，则每来一个任务，会尝试将其添加到任务缓存队列当中，若添加成功，则该任务会等待空闲线程将其取出去执行；若添加失败（一般来说是任务缓存队列已满），则会尝试创建新的线程去执行这个任务；
• 如果当前线程池中的线程数目达到maximumPoolSize，则会采取任务拒绝策略进行处理；
• 如果允许为核心池中的线程设置存活时间，那么核心池中的线程空闲时间超过keepAliveTime，线程会被终止。

4.2.3、任务缓存队列及排队策略

workQueue的类型为BlockingQueue<Runnable>，通常可以取下面三种类型：

• ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小；
• LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX_VALUE；
• synchronousQueue：这个队列比较特殊，它不会保存提交的任务，而是将直接新建一个线程来执行新来的任务。

4.2.4、常见的四种线程

Executors是静态工厂类。通过静态工厂方法返回ExecutorService、ScheduledExecutorService、ThreadFactory 和 Callable 等类的对象。

常见的四种线程有

Executors.newFixedThreadPool

只有核心线程,并且数量固定的,也不会被回收,所有线程都活动时,因为队列没有限制大小,新任务会等待执行.
优点:更快的响应外界请求.

创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

Executors.newCachedThreadPool

只有非核心线程,最大线程数非常大,所有线程都活动时会为新任务创建新线程,否则会利用空闲线程(60s空闲时间,过了就会被回收,所以线程池中有0个线程的可能)处理任务.
优点:任何任务都会被立即执行(任务队列SynchronousQuue相当于一个空集合);比较适合执行大量的耗时较少的任务.

创建一个可缓存线程池。当空闲的线程超过60s内没有用到的话，就会被回收。若无可回收，则新建线程。

    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

**
Executors.singleThreadPoll**

只有一个核心线程,确保所有的任务都在同一线程中按序完成.因此不需要处理线程同步的问题.

创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务

   public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

Executors.ScheduledThreadPol
核心线程数固定,非核心线程（闲着没活干会被立即回收数）没有限制.
优点:执行定时任务以及有固定周期的重复任务

创建一个定长线程池，支持定时及周期性任务执行

    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE,
              DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
              new DelayedWorkQueue());
    }

实例：

public class ThreadPoolText {

    private static final int CORE_POOL_SIZE = 1;
    private static final int MAXIMUM_POOL_SIZE = 2;
    private static final int CAPACITY = 2;

    public static void main(String[] args) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAXIMUM_POOL_SIZE,
                0,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(CAPACITY));
        threadPoolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
        // 新建10个任务，并将它们添加到线程池中。
        for (int i = 0; i < 10; i++) {
            Runnable run = new Thread1("task-" + i);
            threadPoolExecutor.execute(run);
        }
        threadPoolExecutor.shutdown();
    }


    static class Thread1 implements Runnable {
        private String name;

        public Thread1(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            try {
                System.out.println(this.name + " is running.");
                Thread.sleep(200);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

运行结果

task-0 is running.
task-3 is running.
task-8 is running.
task-9 is running.

解释：

核心池的大小是1，最大线程池数是2，keepAliveTime是0空闲线程会被终止，缓存队列是ArrayBlockingQueue，大小是2。拒绝策略是DiscardOldestPolicy。

1. task-0 已运行，（核心池大小是1）
2. task-1、task-2被添加到缓存队列（缓存队列大小是2）
3. task-3时添加失败，缓存队列已满，会尝试创建新的线程去执行这个任务，task-3成功运行，（最大线程池数是2）
4. task-4 ~ task-9 线程池数量达到maximumPoolSize。依次丢弃阻塞队列中旧任务，将被拒绝的任务添加到阻塞队列。最终剩下task-8、task-9（阻塞队列是DiscardOldestPolicy）
5. task-0、task-3执行完再依次去缓存队列的task-8、task-9运行

5、Callable、Future、FutureTask

Callable可以返回执行结果。常用Future搭配使用，Future用于获取结果、查询是否完成、取消

Callable是个接口，包含call()方法,类似于Runnable的run()方法。

public interface Callable<V> {
    V call() throws Exception;
}

Future是个接口，对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。而FutureTask是它的实现类

public interface Future<V> {
    // 试图取消对此任务的执行。
    boolean     cancel(boolean mayInterruptIfRunning)

    // 如果在任务正常完成前将其取消，则返回 true。
    boolean     isCancelled()

    // 如果任务已完成，则返回 true。
    boolean     isDone()

    // 如有必要，等待计算完成，然后获取其结果。
    V           get() throws InterruptedException, ExecutionException;

    // 如有必要，最多等待为使计算完成所给定的时间之后，获取其结果（如果结果可用）。
    V             get(long timeout, TimeUnit unit)
          throws InterruptedException, ExecutionException, TimeoutException;
}

Callable、Future、FutureTask关系图

image.png

RunnableFuture继承Runnable和Future

public interface RunnableFuture<V> extends Runnable, Future<V> {
    void run();
}

FutureTask实现RunnableFuture

public FutureTask(Callable<V> callable) 
public FutureTask(Runnable runnable, V result)

Callable常用ExecutorService搭配使用，ExecutorService的submit方法如下

<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);

实例

public class CallableTest {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Future<Integer> future = executorService.submit(new MyCallable());
        executorService.shutdown();
        System.out.println("isDone " + future.isDone());
        try {
            Integer integer = future.get();
            System.out.println("执行结果 = " + integer);

        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println("isDone " + future.isDone());
        System.out.println("isCancelled " + future.isCancelled());
        System.out.println("主线程在执行任务");

    }

    static class MyCallable implements Callable<Integer> {

        @Override
        public Integer call() throws Exception {
            System.out.println("子线程在进行计算");
            Thread.sleep(3000);
            int sum = 0;
            for (int i = 0; i < 100; i++)
                sum += i;
            return sum;
        }
    }
}

执行结果

isDone false
子线程在进行计算
执行结果 = 4950
isDone true
isCancelled false
主线程在执行任务

可见get()方法会阻塞线程

6、Timer、TimerTask

6.1、Timer和TimerTask

schedule(TimerTask task, long delay) //调度task，延迟delay调度一次
schedule(TimerTask task, Date time) //调度task,在指定time时间调度一次
schedule(TimerTask task, long delay, long period) 
//调度task，在延迟delay调度，每次调度完后，最少等待period（ms）后才开始调度。
schedule(TimerTask task, Date firstTime, long period) 
//第一次调度时间是firstTime，每次调度完后，最少等待period（ms）后才开始调度。

scheduleAtFixedRate(TimerTask task, long delay, long period)
void scheduleAtFixedRate(TimerTask task, Date firstTime,long period)

cancel()    取消单个任务的执行，即将其状态置为CANCELLED
int purge() 会将任务队列中状态为CANCELLED的任务清除，并对最小堆进行重排序。

schedule下次执行时间是：当前时间（任务执行前得到）+时间片

scheduleAtFixedRate 下次执行时间是下个时间片执行时间，优先调度，减少漏掉调度。

实例

Timer timer = new Timer();
timer.schedule(new TimerTask() {
        public void run() {
            System.out.println("abc");
        }
}, new Date() , 1000);

6.2、Timer和ScheduledThreadPoolExecutor的区别

Timer是单线程的，一个定时器不允许同时执行任务。由源码可以看到，Timer内部包装“一个Thread”和“一个Task”队列，这个队列按照一定的方式将任务排队处理，包含的线程在Timer的构造方法调用时被启动，这个Thread的run方法无限循环这个Task队列，若队列为空且没发生cancel操作，此时会一直等待，如果等待完成后，队列还是为空，则认为发生了cancel从而跳出死循环，结束任务；循环中如果发现任务需要执行的时间小于系统时间，则需要执行，那么根据任务的时间片从新计算下次执行时间，若时间片为0代表只执行一次，则直接移除队列即可。

ScheduledThreadPoolExecutor类来实现多线程的定时器。

Timer可以做的事情，通过ScheduledThreadPoolExecutor都可以做到，并且修复了其中Timer的部分设计上的缺陷。并且实现了多线程的执行，相对来说功能是比Timer强大了一个等级。

7、ThreadLocal

用于创建线程的本地变量。

为变量在每个线程创建一个本地副本。每个线程都可以访问自己内部的副本。

即每个线程都有改变量，在线程内部任何地方都可以使用，线程之间不影响，也就不存在现线程安全问题，也不会严重影响线程执行性能

分析源码可以得出：

• 通过ThreadLocal创建的副本是存储在每个线程自己的threadLocals中的
• 一个线程可以有多个ThreadLocal
• 在进行get之前，必须先set，否则会报空指针异常。（如果想在get之前不需要调用set就能正常访问的话，必须重写initialValue()方法。）

常用方法

public T get() { }
public void set(T value) { }
public void remove() { }
protected T initialValue() { }

实例

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;

}

参考：

Java多线程系列目录(共43篇)

随笔分类 - Java并发编程