5.线程池原理剖析&锁的深度化

作者: okhoogh | 来源:发表于2020-09-13 16:09 被阅读0次

线程池

什么是线程池

Java中的线程池是运用场景最多的并发框架，几乎所有需要异步或并发执行任务的程序都可以使用线程池。在开发过程中，合理地使用线程池能够带来3个好处。
第一：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
第二：提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
第三：提高线程的可管理性。线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。但是，要做到合理利用

线程池作用

线程池是为突然大量爆发的线程设计的，通过有限的几个固定线程为大量的操作服务，减少了创建和销毁线程所需的时间，从而提高效率。
如果一个线程的时间非常长，就没必要用线程池了(不是不能作长时间操作，而是不宜。)，况且我们还不能控制线程池中线程的开始、挂起、和中止。

线程池的分类

ThreadPoolExecutor

Java是天生就支持并发的语言，支持并发意味着多线程，线程的频繁创建在高并发及大数据量是非常消耗资源的，因此java提供了线程池。在jdk1.5以前的版本中，线程池的使用是及其简陋的，但是在JDK1.5后，有了很大的改善。

JDK1.5之后加入了java.util.concurrent包，java.util.concurrent包的加入给予开发人员开发并发程序以及解决并发问题很大的帮助。这篇文章主要介绍下并发包下的Executor接口，Executor接口虽然作为一个非常旧的接口（JDK1.5 2004年发布），但是很多程序员对于其中的一些原理还是不熟悉。

Executor框架的最顶层实现是ThreadPoolExecutor类，Executors工厂类中提供的newScheduledThreadPool、newFixedThreadPool、newCachedThreadPool方法其实也只是ThreadPoolExecutor的构造函数参数不同而已。通过传入不同的参数，就可以构造出适用于不同应用场景下的线程池，那么它的底层原理是怎样实现的呢，这篇就来介绍下ThreadPoolExecutor线程池的运行过程。

ThreadPoolExecutor类的一些参数

corePoolSize：核心池的大小。当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中
maximumPoolSize：线程池最大线程数。它表示在线程池中最多能创建多少个线程；
keepAliveTime：表示线程没有任务执行时最多保持多久时间会终止。
unit：参数keepAliveTime的时间单位，有7种取值，在TimeUnit类中有7种静态属性

线程池四种创建方式

Java通过Executors（jdk1.5并发包）提供四种线程池，分别为：

newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
newFixedThreadPool创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
newScheduledThreadPool创建一个定长线程池，支持定时及周期性任务执行。
newSingleThreadExecutor创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

newCachedThreadPool

创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

示例代码如下：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class newCachedThreadPoolDemo {
    public static void main(String[] args) {
        ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();
        for (int i = 1; i <= 10; ++i) {
            int temp = i;
            newCachedThreadPool.execute(() -> System.out.println(Thread.currentThread().getName() 
                                                              + ", i:" + temp));
        }
    }
}

运行结果

总结: 线程池为无限大，当执行第二个任务时第一个任务已经完成，会复用执行第一个任务的线程，而不用每次新建线程。

newFixedThreadPool

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

示例代码如下：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class newFixedThreadPoolDemo {
    public static void main(String[] args) {
        // 1.可缓存的线程池
        ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(3);
        for (int i = 1; i <= 10; ++i) {
            int temp = i;
            newFixedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + ",i:" + temp);
                }
            });
        }
    }
}

运行结果

注意:定长线程池的大小最好根据系统资源进行设置。
如Runtime.getRuntime().availableProcessors()

newScheduledThreadPool

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

示例代码如下：

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class newScheduledThreadPoolDemo {
    public static void main(String[] args) {
        // 3.可定时的线程池
        ScheduledExecutorService newScheduledThreadPool = Executors.newScheduledThreadPool(3);
        for (int i = 1; i <= 10; ++i) {
            int temp = i;
            newScheduledThreadPool.schedule(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + ",i:" + temp);
                }
            }, 3, TimeUnit.SECONDS);
        }
    }
}

运行结果

延迟3秒执行。

newSingleThreadExecutor

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

示例代码如下：

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class newSingleThreadPoolDemo {
    public static void main(String[] args) {
        // 4.定长的线程池
        ExecutorService newSingleThreadExecutor = Executors.newSingleThreadExecutor();
        for (int i = 1; i <= 10; ++i) {
            int temp = i;
            newSingleThreadExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + ",i:" + temp);
                }
            });
        }
    }
}

运行结果

注意: 结果依次输出，相当于顺序执行各个任务。

newSingleThreadExecutor.shutdown();等所有线程执行完之后在关掉线程池

线程池原理剖析

提交一个任务到线程池中，线程池的处理流程如下：

判断线程池里的核心线程是否都在执行任务，如果不是（核心线程空闲或者还有核心线程没有被创建）则创建一个新的工作线程来执行任务。如果核心线程都在执行任务，则进入下个流程。
线程池判断工作队列是否已满，如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。
判断线程池里的线程是否都处于工作状态，如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务。

合理配置线程池

要想合理的配置线程池，就必须首先分析任务特性，可以从以下几个角度来进行分析：

任务的性质：CPU密集型任务，IO密集型任务和混合型任务。
任务的优先级：高，中和低。
任务的执行时间：长，中和短。
任务的依赖性：是否依赖其他系统资源，如数据库连接。

任务性质不同的任务可以用不同规模的线程池分开处理。CPU密集型任务配置尽可能少的线程数量，如配置Ncpu+1个线程的线程池。IO密集型任务则由于需要等待IO操作，线程并不是一直在执行任务，则配置尽可能多的线程，2xNcpu。 混合型的任务，如果可以拆分，则将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐率要高于串行执行的吞吐率，如果这两个任务执行时间相差太大，则没必要进行分解。我们可以通过Runtime.getRuntime().availableProcessors()方法获得当前设备的CPU个数。
优先级不同的任务可以使用优先级队列PriorityBlockingQueue来处理。它可以让优先级高的任务先得到执行，需要注意的是如果一直有优先级高的任务提交到队列里，那么优先级低的任务可能永远不能执行。
执行时间不同的任务可以交给不同规模的线程池来处理，或者也可以使用优先级队列，让执行时间短的任务先执行。
依赖数据库连接池的任务，因为线程提交SQL后需要等待数据库返回结果，如果等待的时间越长CPU空闲时间就越长，那么线程数应该设置越大，这样才能更好的利用CPU。

CPU密集型时，任务可以少配置线程数，大概和机器的cpu核数相当，这样可以使得每个线程都在执行任务

IO密集型时，大部分线程都阻塞，故需要多配置线程数，2*cpu核数

某些进程花费了绝大多数时间在计算上，而其他则在等待I/O上花费了大多是时间，
前者称为计算密集型（CPU密集型）computer-bound，后者称为I/O密集型，I/O-bound。

Java锁的深度化

悲观锁、乐观锁、排他锁

场景：
当多个请求同时操作数据库时，首先将订单状态改为已支付，在金额加上200，在同时并发场景查询条件下，会造成重复通知。

悲观锁与乐观锁

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

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

重入锁

锁作为并发共享数据，保证一致性的工具，在JAVA平台有多种实现(如 synchronized 和 ReentrantLock等等 ) 。这些已经写好提供的锁为我们开发提供了便利。
重入锁，也叫做递归锁，指的是同一线程中外层函数获得锁之后，内层递归函数仍然有获取该锁的代码，但不受影响。

在JAVA环境下 ReentrantLock 和synchronized 都是可重入锁

`synchronized`示例代码如下：

public class ReentrantLockDemo1 implements Runnable {
    public synchronized void get() {
        System.out.println("name:" + Thread.currentThread().getName() + " get();");
        set();
    }

    public synchronized void set() {
        System.out.println("name:" + Thread.currentThread().getName() + " set();");
    }

    @Override
    public void run() {
        get();
    }

    public static void main(String[] args) {
        ReentrantLockDemo1 demo1 = new ReentrantLockDemo1();
        new Thread(demo1).start();
        new Thread(demo1).start();
        new Thread(demo1).start();
        new Thread(demo1).start();
    }
}

运行结果

`ReentrantLock`示例代码如下：

import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockDemo2 extends Thread {
    ReentrantLock lock = new ReentrantLock();

    public void get() {
        lock.lock();
        System.out.println("name:" + Thread.currentThread().getName() + " get();");
        set();
        lock.unlock();
    }

    public void set() {
        lock.lock();
        System.out.println("name:" + Thread.currentThread().getName() + " set();");
        lock.unlock();
    }

    @Override
    public void run() {
        get();
    }

    public static void main(String[] args) {
        ReentrantLockDemo2 demo2 = new ReentrantLockDemo2();
        new Thread(demo2).start();
        new Thread(demo2).start();
        new Thread(demo2).start();
    }
}

运行结果同上

读写锁

相比Java中的锁Lock实现，读写锁更复杂一些。假设你的程序中涉及到对一些共享资源的读和写操作，且写操作没有读操作那么频繁。在没有写操作的时候，两个线程同时读一个资源没有任何问题，所以应该允许多个线程能在同时读取共享资源。但是如果有一个线程想去写这些共享资源，就不应该再有其它线程对该资源进行读或写（也就是说：读-读能共存，读-写不能共存，写-写不能共存）。这就需要一个读/写锁来解决这个问题。Java5在java.util.concurrent包中已经包含了读写锁。

`ReentrantReadWriteLock`示例代码如下：

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ReadWriteLock {
    private static Map<String, Object> map = new HashMap<String, Object>();
    private static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
    private static Lock r = rwl.readLock();
    private static Lock w = rwl.writeLock();

    // 获取一个key对应的value
    public static final Object get(String key) {
        r.lock();
        try {
            System.out.println("读key:" + key + " 开始");
            Thread.sleep(100);
            Object object = map.get(key);
            System.out.println("读key:" + key + " 结束");
            return object;
        } catch (InterruptedException e) {
        } finally {
            r.unlock();
        }
        return key;
    }

    // 设置key对应的value，并返回旧有的value
    public static final Object put(String key, Object value) {
        w.lock();
        try {
            System.out.println("写key:" + key + ",value:" + value + "开始");
            Thread.sleep(100);
            Object object = map.put(key, value);
            System.out.println("写key:" + key + ",value:" + value + "结束");
            return object;
        } catch (InterruptedException e) {
        } finally {
            w.unlock();
        }
        return value;
    }

    // 清空所有的内容
    public static final void clear() {
        w.lock();
        try {
            map.clear();
        } finally {
            w.unlock();
        }
    }

    public static void main(String[] args) {
        new Thread(() -> {
            for (int i = 1; i <= 5; i++) {
                ReadWriteLock.put(i + "", i + "");
            }

        }).start();
        new Thread(() -> {
            for (int i = 1; i <= 5; i++) {
                ReadWriteLock.get(i + "");
            }
        }).start();
    }
}

运行结果同上

CAS无锁机制

与锁相比，使用比较交换（下文简称CAS）会使程序看起来更加复杂一些。但由于其非阻塞性，它对死锁问题天生免疫，并且，线程间的相互影响也远远比基于锁的方式要小。更为重要的是，使用无锁的方式完全没有锁竞争带来的系统开销，也没有线程间频繁调度带来的开销，因此，它要比基于锁的方式拥有更优越的性能。
无锁的好处：
第一，在高并发的情况下，它比有锁的程序拥有更好的性能；
第二，它天生就是死锁免疫的。
就凭借这两个优势，就值得我们冒险尝试使用无锁的并发。
CAS算法的过程是这样：它包含三个参数CAS(V, E, N): V表示要更新的变量，E表示预期值，N表示新值。仅当V值等于E值时，才会将V的值设为N，如果V值和E值不同，则说明已经有其他线程做了更新，则当前线程什么都不做。最后，CAS返回当前V的真实值。
CAS操作是抱着乐观的态度进行的，它总是认为自己可以成功完成操作。当多个线程同时使用CAS操作一个变量时，只有一个会胜出，并成功更新，其余均会失败。失败的线程不会被挂起，仅是被告知失败，并且允许再次尝试，当然也允许失败的线程放弃操作。基于这样的原理，CAS操作即使没有锁，也可以发现其他线程对当前线程的干扰，并进行恰当的处理。
简单地说，CAS需要你额外给出一个期望值，也就是你认为这个变量现在应该是什么样子的。如果变量不是你想象的那样，那说明它已经被别人修改过了。你就重新读取，再次尝试修改就好了。
在硬件层面，大部分的现代处理器都已经支持原子化的CAS指令。在JDK 5.0以后，虚拟机便可以使用这个指令来实现并发操作和并发数据结构，并且，这种操作在虚拟机中可以说是无处不在。

自旋锁

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

import java.util.concurrent.atomic.AtomicReference;

class SpinLock {
    private AtomicReference<Thread> sign = new AtomicReference<>();
    public void lock() {
        Thread current = Thread.currentThread();
        while (!sign.compareAndSet(null, current)) {
        }
    }

    public void unlock() {
        Thread current = Thread.currentThread();
        sign.compareAndSet(current, null);
    }
}

public class SpinLockDemo implements Runnable {
    static int sum;
    private SpinLock lock;

    public SpinLockDemo(SpinLock lock) {
        this.lock = lock;
    }

    @Override
    public void run() {
        this.lock.lock();
//        this.lock.lock();
        sum++;
        this.lock.unlock();
//        this.lock.unlock();
    }
    
    public static void main(String[] args) throws InterruptedException {
        SpinLock lock = new SpinLock();
        for (int i = 0; i < 100; i++) {
            SpinLockDemo test = new SpinLockDemo(lock);
            Thread t = new Thread(test);
            t.start();
        }
        Thread.currentThread().sleep(1000);
        System.out.println(sum);
    }
}

当一个线程调用这个不可重入的自旋锁去加锁的时候没问题，当再次调用lock()的时候，因为自旋锁的持有引用已经不为空了，该线程对象会误认为是别人的线程持有了自旋锁
使用了CAS原子操作，lock函数将owner设置为当前线程，并且预测原来的值为空。unlock函数将owner设置为null，并且预测值为当前线程。
当有第二个线程调用lock操作时由于owner值不为空，导致循环一直被执行，直至第一个线程调用unlock函数将owner设置为null，第二个线程才能进入临界区。

由于自旋锁只是将当前线程不停地执行循环体，不进行线程状态的改变，所以响应速度更快。但当线程数不停增加时，性能下降明显，因为每个线程都需要执行，占用CPU时间。如果线程竞争不激烈，并且保持锁的时间段。适合使用自旋锁。

5.线程池原理剖析&锁的深度化

线程池

什么是线程池

线程池作用

线程池的分类

ThreadPoolExecutor

线程池四种创建方式

newCachedThreadPool

示例代码如下：

运行结果

newFixedThreadPool

示例代码如下：

运行结果

newScheduledThreadPool

示例代码如下：

运行结果

newSingleThreadExecutor

示例代码如下：

运行结果

线程池原理剖析

合理配置线程池

Java锁的深度化

悲观锁、乐观锁、排他锁

悲观锁与乐观锁

重入锁

`synchronized`示例代码如下：

运行结果

`ReentrantLock`示例代码如下：

运行结果同上

读写锁

`ReentrantReadWriteLock`示例代码如下：

运行结果同上

CAS无锁机制

自旋锁

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

5.线程池原理剖析&锁的深度化

线程池

什么是线程池

线程池作用

线程池的分类

ThreadPoolExecutor

线程池四种创建方式

newCachedThreadPool

示例代码如下：

运行结果

newFixedThreadPool

示例代码如下：

运行结果

newScheduledThreadPool

示例代码如下：

运行结果

newSingleThreadExecutor

示例代码如下：

运行结果

线程池原理剖析

合理配置线程池

Java锁的深度化

悲观锁、乐观锁、排他锁

悲观锁与乐观锁

重入锁

synchronized示例代码如下：

运行结果

ReentrantLock示例代码如下：

运行结果同上

读写锁

ReentrantReadWriteLock示例代码如下：

运行结果同上

CAS无锁机制

自旋锁

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

`synchronized`示例代码如下：

`ReentrantLock`示例代码如下：

`ReentrantReadWriteLock`示例代码如下：