线程池原理分析

CountDownLatch（计数器）

CountDownLatch 位于并发包下，利用它可以完成类似于计数器的功能，如果线程 A 需要等待其他 n 个线程执行完毕后才能执行，此时就可以利用 CountDownLatch 来实现这个功能，CountDownLatch 是通过一个计数器来实现的，计数器的初始值为线程数量，每当一个线程完成了自己的任务后，计数器的值就会减1，当计数器的值为0时，表示所有线程已经执行完毕，等待线程就可以恢复。

创新互联专注于辽中企业网站建设,响应式网站,商城网站建设。辽中网站建设公司,为辽中等地区提供建站服务。全流程按需求定制制作，专业设计，全程项目跟踪，创新互联专业和态度为您提供的服务

package com.kernel;

import java.util.concurrent.CountDownLatch;

public class Test001 {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(2);
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + ",子线程开始执行...");
                countDownLatch.countDown();
                System.out.println(Thread.currentThread().getName() + ",子线程结束执行...");

            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName() + ",子线程开始执行...");
                countDownLatch.countDown();
                System.out.println(Thread.currentThread().getName() + ",子线程结束执行...");

            }
        }).start();
        countDownLatch.await();
        System.out.println("其他线程执行完毕");
        System.out.println("等待线程执行。。。");
    }
}

CyclicBarrier（屏障）

CyclicBarrier 在初始化时会传入一个数量，它会记录调用了 await 方法的线程数，只有这个线程数和创建该对象时提供的数量相同时，所有进入线程等待的线程才会被重新唤醒继续执行。

顾名思义，它就像一个屏幕，人来全了才能一块儿过屏障。

CyclicBarrier 还可以提供一个可以传入 Runable 对象的构造，该线程将在一起通过屏障后所有线程唤醒之前被唤醒

package com.kernel;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

class Write extends Thread {
    private CyclicBarrier cyclicBarrier;

    public Write(CyclicBarrier cyclicBarrier) {
        this.cyclicBarrier = cyclicBarrier;
    }

    @Override
    public void run() {
        System.out.println("线程" + Thread.currentThread().getName() + ",正在写入数据");
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("线程" + Thread.currentThread().getName() + ",写入数据成功.....");
        try {
            cyclicBarrier.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        System.out.println("所有线程执行完毕");
    }
}

public class Test002 {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(5);
        for (int i = 0; i < 5; i++) {
            Write write = new Write(cyclicBarrier);
            write.start();
        }
    }
}

Semaphore（信号量）

Semaphore 是一种基于计数的信号量，创建时可以指定一个值，这个值规定了有多少个线程并发执行，执行前申请，执行完毕后归还，超过那个值后，线程申请信号将会被阻塞，知道有其他占有信号的线程执行完成归还信号

Semaphore 可以用来构建一些对象池，资源池之类的，比如数据库连接池

我们也可以创建计数为1的 Semaphore，将其作为一种类似互斥锁的机制，这也叫二元信号量，表示两种互斥状态

package com.kernel;

import java.util.Random;
import java.util.concurrent.Semaphore;

public class Test003 extends Thread {
    private String name;
    private Semaphore windows;

    public Test003(String name, Semaphore windows) {
        this.name = name;
        this.windows = windows;
    }

    @Override
    public void run() {
        int availablePermits = windows.availablePermits();
        if (availablePermits > 0) {
            System.out.println(name + ":终于轮到我了");
        } else {
            System.out.println(name + ":**，能不能快点");
        }
        try {
            windows.acquire();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(name + ":我要XXX，剩下窗口");
        try {
            Thread.sleep(new Random().nextInt(1000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(name + ":我买完了");
        windows.release();

    }

    public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(3);
        for (int i = 0; i < 5; i++) {
            Test003 test003 = new Test003("第" + i + "个人", semaphore);
            test003.start();
        }
    }
}

并发队列

ArrayDeque（数组双端队列）
PriorityQueue（优先级队列）
ConcurrentLinkedQueue（基于链表的并发队列）
DelayQueue（延期阻塞队列，阻塞队列实现了BlockingQueue接口）
ArrayBlockingQueue（基于数组的并发阻塞队列）
LinkedBlockingQueue（基于链表的FIFO阻塞队列）
LinkedBlockingDeque（基于链表的FIFO双端阻塞队列）
PriorityBlockingQueue（带优先级的×××阻塞队列）
SynchronousQueue（并发同步阻塞队列）

ConcurrentLinkedDeque（非阻塞式队列）

是一个×××队列，性能高于阻塞式队列，是一个基于链表实现的线程安全队列，不允许 null

add 和 offer 是加入元素的方法，两者之间没有区别

poll 从队列中取出并删除元素

peek 查看队列头元素，不删除

BlockingQueue（阻塞式队列）

是一个有界队列，阻塞队列常用语生产消费者场景

以下情况会产生阻塞：

队列元素满，还往里面存元素

队列元素空，还想从队列中拿元素

基于阻塞队列的生产者消费者模型

package com.kernel;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

class ProducerThread implements Runnable {
    private BlockingQueue blockingQueue;
    private AtomicInteger atomicInteger = new AtomicInteger();
    private volatile boolean flag = true;

    public ProducerThread(BlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {
        System.out.println("###生产者线程已经启动###");
        while (flag) {
            String data = atomicInteger.incrementAndGet() + "";
            try {
                boolean offer = blockingQueue.offer(data, 2, TimeUnit.SECONDS);
                if (offer) {
                    System.out.println(Thread.currentThread().getName() + ", 生产队列" + data + "成功");
                } else {
                    System.out.println(Thread.currentThread().getName() + ", 生产队列" + data + "失败");
                }
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("生产者线程结束");
    }

    public void stop() {
        this.flag = false;
    }
}

class CustomerThread implements Runnable {

    private BlockingQueue blockingQueue;
    private volatile boolean flag = true;

    public CustomerThread(BlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    @Override
    public void run() {
        System.out.println("###消费者线程已经启动###");
        while (flag) {
            String data = null;
            try {
                data = blockingQueue.poll(2, TimeUnit.SECONDS);
                if (data == null) {
                    flag = false;
                    System.out.println("消费者超过2秒时间未获取到消息.");
                    return;
                }
                System.out.println("消费者获取到队列信息成功,data:" + data);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("消费者进程结束");
    }
}

public class Test006 {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
        ProducerThread producerThread = new ProducerThread(blockingQueue);
        CustomerThread customerThread = new CustomerThread(blockingQueue);
        new Thread(producerThread).start();
        new Thread(customerThread).start();
        Thread.sleep(1000 * 10);
        producerThread.stop();
    }
}

线程池介绍

线程池其实就是一个可以容纳线程的容器，其中的线程可以反复利用，节省了反复创建、销毁线程的消耗

线程池有什么优点？

降低资源消耗：重复利用已经创建好的线程而节约反复创建、销毁线程的消耗

提高响应速度：众所周期，创建线程不是立马可以使用的，创建好线程之后进入就绪状态，需要经过 CPU 的调度才能进入运行状态，而利用线程池，只要任务来到，线程池有空闲线程，就可以立即作业

提高线程管理性：线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还可以降低系统稳定性，使用线程池可以进行统一分配、调优和监控

线程池的分类：

newCachedThreadPool

创建一个可缓存线程池，可反复回收利用，若任务数大于当然线程数，则继续创建线程

public class Test008 {
    public static void main(String[] args) {
        // 可缓存线程池（可重复利用）无限大
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 100; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName());
                }
            });
        }
    }
}

newFixedThreadPool

创建一个定长线程，超出线程存放在队列中

public class Test009 {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 10; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName());
                }
            });
        }
    }
}

newScheduledThreadPool

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

public class Test010 {
    public static void main(String[] args) {
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
        for (int i = 0; i < 10; i++) {
            scheduledExecutorService.schedule(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName());
                }
                // 表示延迟3秒执行
            },3, TimeUnit.SECONDS);
        }
    }
}

newSingleThreadExecutor

创建一个单线程线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序（FIFO、LIFO、优先级）执行

public class Test011 {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 10; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName());
                }
            });
        }
        executorService.shutdown();
    }
}

线程池源码分析

corePoolSize 核心线程数（实际运行线程数）
maximumPoolSize 最大线程数（最多可以创建多少个线程）
keepAliveTime 线程空闲超时时间
unit 时间单位
workQueue 缓存队列

提交一个任务到线程池中去，首先判断当前线程数是都小于 corePoolSize，如果小于 corePoolSize，则创建一个新线程来执行任务

如果当前线程数等于corePoolSize，再来任务的话就会将任务添加到缓存队列中

如果缓存队列已满，在判断当前线程是否小于 maximumPoolSize

如果小于 maximumPoolSize，创建线程执行任务，否则，采取任务拒绝策略进行处理

如果当前线程数大于 corePoolSize，并且某线程空闲时间大于 keepAliveTime，线程被终止，直到线程池中的线程数目不大于corePoolSize

如果允许为核心池中的线程设置存活时间，那么核心池中的线程空闲时间超过 keepAliveTime，线程也会被终止

合理配置线程池

IO 密集型：即该任务需要大量的IO，即大量的阻塞，在单线程上运行IO密集型的任务会导致浪费大量的 CPU 运算能力浪费在等待，所以在 IO 密集型任务中使用多线程可以大大的加速程序运行，即时在单核CPU上，这种加速主要就是利用了被浪费掉的阻塞时间，一般 IO 密集型任务线程设置为 2*核心数+1

CPU 密集型：该任务需要进行大量计算，没有 IO 阻塞，CPU 一直在全速运行，CPU 密集任务只有在真正的多核CPU上才可能得到加速（通过多线程），而在单核CPU上，无论你开几个模拟的多线程，该任务都不可能得到加速，因为CPU总的运算能力就那些，一般 CPU 密集型任务线程设置为核心数+1

Callable

Java中，创建线程一般有两种方式，就是继承 Thread 或者实现 Runable 接口，这两种方式的缺点是在线程任务执行完毕后，无法获得执行结果，所以一般使用共享变量或者共享存储区以及线程通信的方式获得结果，Java 中也提供了使用 Callable 和 Future 来实现获取任务结果的操作，Callable 用来执行任务，产生结果，而 Future 用来获得结果

Future 常用方法

V get() 获取异步执行的结果，如果没有结果可用，此方法会阻塞直到异步计算完成

V get(Long timeout, Timeunit unit) 获取异步执行结果，如果没有结果可用，此方法会阻塞，但是会有时间限制，如果阻塞时间超过设定的timeout时间，该方法将抛出异常

boolean isDone() 如果任务执行结束，无论是正常结束或是中途取消还是发生异常，都返回true

boolean isCanceller() 如果任务完成前被取消，则返回true

boolean cancel(boolean mayInterruptIfRunning) 如果任务还没开始就执行该方法将返回 false，如果任务执行过程中调用 cancel(true) 将以中断执行任务的方式试图停止任务，如果停止成功，返回 true，如果执行 cancel(false) 不会对执行的任务产生影响，此时返回 false，当任务完成后调用该方法将返回 false，参数表示是否中断执行进程

Future 模式

去除主函数的等待时间，使得原本需要等待的时间可以处理其他业务，对于多线程，如果线程 A 要等待线程 B 的结果，那么线程 A 没必要等待 B，直到 B 有结果，可以先拿到一个未来的 Future，等 B 有结果是再取真实的结果

模拟 Future

Data

public interface Data {
    // 获取子线程执行结果
    public String getRequest() throws InterruptedException;
}

FutureData

public class FutureData implements Data {
    private boolean flag = false;
    private RealData realData;

    public synchronized void setRealData(RealData realData) {
        if (flag)
            return;
        this.realData = realData;
        flag = true;
        // 唤醒
        notify();
    }

    @Override
    public synchronized String getRequest() throws InterruptedException {
        while (!flag) {
            // 等待
            wait();
        }
        // 返回结果
        return realData.getRequest();
    }
}

RealData

public class RealData implements Data {
    private String result;

    public RealData(String data) throws InterruptedException {
        System.out.println("正在下载" + data);
        Thread.sleep(5000);
        System.out.println("下载完毕！");
        result = "author:kernel";
    }

    @Override
    public String getRequest() {
        return result;
    }
}

FutureClient

public class FutureClient {
    public Data submit(String requestData) {
        FutureData futureData = new FutureData();
        new Thread(new Runnable() {
            @Override
            public void run() {
                RealData realData = null;
                try {
                    realData = new RealData(requestData);
                    futureData.setRealData(realData);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        return futureData;
    }
}

测试

public class Test002 {
    public static void main(String[] args) throws InterruptedException {
        FutureClient futureClient = new FutureClient();
        Data request = futureClient.submit("请求参数");
        System.out.println("请求发送成功");
        // 主线程该干嘛干嘛去
        System.out.println("执行其他任务");
        // 获取其他线程的结果
        String result = request.getRequest();
        System.out.println("获取到结果" + result);
    }
}

执行流程：

首先创建一个向 FutureClient 发送请求，然后 realData 执行下载任务，将结果封装起来，然后获取结果函数时刻监测线程是否拿到结果，如果拿到了，就返回，如果没有拿到，就一直阻塞，设置结果的函数拿到结果会立即唤醒返回结果的函数

重入锁

重入锁，也叫递归锁，指的是同一线程外层函数获得锁之后，内存函数仍有获取该锁的代码，但不受影响

ReentrantLock（显式锁、轻量级锁）和 Synchronized（内置锁、重量级锁）都是可重入锁

读写锁

程序中涉及一些对共享变量的读写操作时，在没有写操作时，多个线程同时读取是没有任何问题的，如果有一个线程正在进行写操作，其他线程就不应该对其进行读或写操作了

读写不同共存，读读可以共存，写写不能共存

public class Test001 {
    private volatile Map cache = new HashMap();
    private ReentrantReadWriteLock reentrantReadWriteLock = new ReentrantReadWriteLock();
    private ReadLock readLock = reentrantReadWriteLock.readLock();
    private WriteLock writeLock = reentrantReadWriteLock.writeLock();

    public void put(String key, String value) {
        try {
            writeLock.lock();
            System.out.println("写入put方法key：" + key + ",value" + value + ",开始");
            Thread.sleep(1000);
            cache.put(key, value);
            System.out.println("写入put方法key：" + key + ",value" + value + ",结束");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            writeLock.unlock();
        }
    }

    public String get(String key) {
        try {
            readLock.lock();
            System.out.println("读取key："+ key + ",开始");
            Thread.sleep(1000);
            String value = (String) cache.get(key);
            System.out.println("读取key："+ key + ",结束");
            return value;
        } catch (InterruptedException e) {
            e.printStackTrace();
            return null;
        } finally {
            readLock.unlock();
        }
    }

    public static void main(String[] args) {
        Test001 test001 = new Test001();
        Thread t1 = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++) {
                    test001.put("i", i + "");
                }
            }
        });
        t1.start();
        Thread t2 = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++) {
                    test001.get(i + "");
                }
            }
        });
        t2.start();
    }
}

悲观锁

悲观锁悲观的认为每一次操作都会造成更新丢失问题，在每次查询时加上排他锁，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会阻塞直到它拿到锁，传统的关系型数据库里边就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁

乐观锁

乐观锁会乐观的认为每次查询都不会造成更新丢失，利用版本字段控制

首先通过条件查询出版本号，然后更新的时候判断当前版本号是否和之前版本号一致，如果一致，证明没人修改，直接更新，否则，重新查询以便在进行更新

自旋锁

自旋锁是采用让当前线程不停地的在循环体内执行实现的，当循环的条件被其他线程改变时才能进入临界区

公平锁和非公平锁

公平锁：新进程发出请求，如果此时一个线程正持有锁，或有其他线程正在等待队列中等待这个锁，那么新的线程将被放入到队列中被挂起

非公平锁：新进程发出请求，如果此时一个线程正持有锁，新的线程将被放入到队列中被挂起，但如果发出请求的同时该锁变成可用状态，那么这个线程会跳过队列中所有的等待线程而获得锁

CAS 无锁机制（利用原子类）

与锁相比，使用 CAS 无锁机制会使程序看起来复杂一些，但由于其非阻塞性，是不会发生死锁现象的，而且线程间相互影响也比锁要小的多，使用 CAS 完全没有锁之间竞争带来的系统开销，也没有线程间频繁调用带来的开销

CAS 原理

CAS 包括三个参数，分别是 V（表示要更新的变量、主内存的值）、E（表示预期值、本地内存的值）、N（新值），如果 V = E，那么将 V 的值设置成 N，如果 V != E，说明其他线程修改了，则当前线程什么都不做，最后，CAS返回当前V的真实值

CAS 和乐观锁很相似，都是抱着乐观的心态去处理，多个线程同时使用 CAS 操作，只有一个能胜出，其他都失败，失败的线程不会挂起，仅告知失败，并且允许重新尝试，当然也允许放弃尝试，基于这样的原理，CAS操作即使没有锁，也可以发现其他线程对当前线程的干扰，并进行恰当的处理

简单地说，CAS需要你额外给出一个期望值，也就是你认为这个变量现在应该是什么样子的，如果变量不是你想象的那样，那说明它已经被别人修改过了。你就重新读取，再次尝试修改就好了

优缺点

优点：

在高并发的情况下，它比有锁的程序拥有更好的性能

死锁免疫

缺点：

CAS存在一个很明显的问题，即 ABA 问题

果在这段期间曾经被改成 B，然后又改回 A，那 CAS 操作就会误认为它从来没有被修改过，针对这种情况，并发包中提供了一个带有标记的原子引用类 AtomicStampedReference，它可以通过控制变量值的版本来保证 CAS 的正确性

原子类

Java 中的原子操作类大致可以分为4类：原子更新基本类型、原子更新数组类型、原子更新引用类型、原子更新属性类型，这些原子类中都是用了无锁的概念，有的地方直接使用CAS操作的线程安全的类型
AtomicBoolean

AtomicInteger

AtomicLong

AtomicReference

。。。

当前标题：线程池原理分析
转载注明：http://myzitong.com/article/iggsdp.html