正文

Java并发编程核心概念一览

IP云
IP云V管理员/4 阅读
06/19
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

作者博客地址 https://muggle.javaboy.org。

并行相关概念

同步和异步

同步和异步通常来形容一次方法的调用。同步方法一旦开始，调用者必须等到方法结束才能执行后续动作；异步方法则是在调用该方法后不必等到该方法执行完就能执行后面的代码，该方法会在另一个线程异步执行，异步方法总是伴随着回调，通过回调来获得异步方法的执行结果。

并发和并行

很多人都将并发与并行混淆在一起，它们虽然都可以表示两个或者多个任务一起执行，但执行过程上是有区别的。并发是多个任务交替执行，多任务之间还是串行的；而并行是多个任务同时执行，和并发有本质区别。

对计算机而言，如果系统内只有一个 CPU ，而使用多进程或者多线程执行任务，那么这种情况下多线程或者多进程就是并发执行，并行只可能出现在多核系统中。当然，对 Java 程序而言，我们不必去关心程序是并行还是并发。

临界区

临界区表示的是多个线程共享但同时只能有一个线程使用它的资源。在并行程序中临界区资源是受保护的，必须确保同一时刻只有一个线程能使用它。

阻塞

如果一个线程占有了临界区的资源，其他需要使用这个临界区资源的线程必须在这个临界区进行等待(线程被挂起)，这种情况就是发生了阻塞(线程停滞不前)。

死锁\饥饿\活锁

死锁就是多个线程需要其他线程的资源才能释放它所拥有的资源，而其他线程释放这个线程需要的资源必须先获得这个线程所拥有的资源，这样造成了矛盾无法解开；如图1情形就是发生死锁现象：

图1：生活中的死锁现象

活锁就是两个线程互相谦让资源，结果就是谁也拿不到资源导致活锁；就好比过马路，行人给车让道，车又给行人让道，结果就是车和行人都停在那不走。

饥饿就是，某个线程优先级特别低老是拿不到资源，导致这个线程一直无法执行。

并发级别

并发级别分为阻塞，无饥饿，无障碍，无锁，无等待几个级别；根据名字我们也能大概猜出这几个级别对应的什么情形；阻塞，无饥饿和无锁都好理解；我们说一下无障碍和无等待；

无障碍：无障碍级别默认各个线程不会发生冲突，不会互相抢占资源，一旦抢占资源就认为线程发生错误，进行回滚。

无等待：无等待是在无锁上的进一步优化，限制每个线程完成任务的步数。

并行的两个定理

加速比：加速比=优化前系统耗时/优化后系统耗时

Amdahl 定理： 加速比=1/[F+(1-F)/n] 其中 n 表示处理器个数 ，F是程序中只能串行执行的比例(串行率)；由公式可知，想要以最小投入，得到最高加速比即 F+(1-F)/n 取到最小值，F 和 n 都对结果有很大影响，在深入研究就是数学问题了。

Gustafson 定律： 加速比=n-F(n-1)，这两定律区别不大，都体现了单纯的减少串行率，或者单纯的加 CPU 都无法得到最优解。

Java 中的并行基础

原子性，可见性，有序性

原子性指的是一个操作是不可中断的，要么成功要么失败，不会被其他线程所干扰；比如 int=1 ,这一操作在 cpu 中分为好几个指令，但对程序而言这几个指令是一体的，只有可能执行成功或者失败，不可能发生只执行了一半的操作；对不同 CPU 而言保证原子性的的实现方式各有不同，就英特尔 CPU 而言是使用一个 lock 指令来保证的。

可见性指某一线程改变某一共享变量，其他线程未必会马上知道。

有序性指对一个操作而言指令是按一定顺序执行的，但编译器为了提高程序执行的速度，会重排程序指令；cpu在执行指令的时候采用的是流水线的形式，上一个指令和下一个指令差一个工步。比如A指令分三个工步：

  1. 操作内存a；

  2. 操作内存b；

  3. 操作内存c；

现假设有个指令 B 操作流程和 A 一样，那么先执行指令 A 再执行指令 B 时间全利用上了，中间没有停顿等待；但如果有三个这样的指令在流水线上执行： a>b>c ， b>e>c ， c>e>a ；这样的指令顺序就会发生等待降低了 CPU 的效率，编译器为了避免这种事情发生，会适当优化指令的顺序进行重排。

volatile关键字

volatile 关键字在 Java 中的作用是保证变量的可见性和防止指令重排。

线程的相关操作

创建线程有三种方法

  • 继承Thread类创建线程

  • 实现Runnable接口创建线程

  • 使用Callable和Future创建线程

终止线程的方法

终止线程可调用 stop() 方法，但这个方法是被废弃不建议使用的，因为强制终止一个线程会引起数据的不一致问题。比如一个线程数据写到一半被终止了，释放了锁，其他线程拿到锁继续写数据，结果导致数据发生了错误。终止线程比较好的方法是“让程序自己终止”，比如定义一个标识符，当标识符为 true 的时候直让程序走到终点，这样就能达到“自己终止”的目的。

线程的中断等待和通知

interrupt() 方法可以中断当前程序，object.wait() 方法让线程进入等待队列，object.notify() 随机唤醒等待队列的一个线程， object.notifyAll() 唤醒等待队列的所有线程。object.wait() 必须在 synchronzied 语句中调用；执行wait、notify 方法必须获得对象的监视器，执行结束后释放监视器供其他线程获取。

join

join() 方法功能是等待其他线程“加入”，可以理解为将某个线程并为自己的子线程，等子线程走完或者等子线程走规定的时间，主线程才往下走；join 的本质是调用调用线程对象的 wait 方法，当我们执行 wait 或者 notify 方法不应该获取线程对象的监听器，因为可能会影响到其他线程的 join。

yield

yield 是线程的“谦让”机制，可以理解为当线程抢到 cpu 资源时，放弃这次资源重新抢占，yield() 是 Thread 里的一个静态方法。

线程组

如果一个多线程系统线程数量众多而且分工明确，那么可以使用线程组来分类。

  1. public void contextLoads() {

  2. ThreadGroup testGroup=new ThreadGroup("testGroup");

  3. Thread a = new Thread(testGroup, new MyRunnable(), "a");

  4. Thread b = new Thread(testGroup, new MyRunnable(), "b");

  5. a.start();

  6. b.start();

  7. int i = testGroup.activeCount();

  8. }

  9. class MyRunnable implements Runnable{

  10. @Override

  11. public void run() {

  12. System.out.println("test");

  13. }

  14. }

图示代码创建了一个 testGroup 线程组。

守护线程

守护线程是一种特殊线程，它类似 Java 中的异常系统，主要是概念上的分类，与之对应的是用户线程。它功能应该是在后台完成一些系统性的服务；设置一个线程为守护线程应该在线程 start 之前 setDaemon()。

线程优先级

Java 中线程可以有自己的优先级，优先级高的更有优势抢占资源；线程优先级高的不一定能抢占到资源，只是一个概率问题，而对应优先级低的线程可能会发生饥饿。

在 Java 中使用1到10表示线程的优先级，使用setPriority()方法来进行设置，数字越大代表优先级越高。

Java 线程锁

以下分类是从多个同角度来划分，而不是以某一标准来划分，请注意：

  • 阻塞锁：当一个线程获得锁，其他线程就会被阻塞挂起，直到抢占到锁才继续执行，这样会导致 CPU 切换上下文，切换上下文对 CPU 而言是很耗费时间的。

  • 非阻塞锁：当一个线程获得锁，其他线程直接跳过锁资源相关的代码继续执行，就是非阻塞锁。

  • 自旋锁：当一个线程获得锁，其他线程则在不停进行空循环，直到抢到锁，这样做的好处是避免了上下文切换。

  • 可重入锁：也叫做递归锁，当一个线程获得该锁后，可以多次进入该锁所同步着的代码块。

  • 互斥锁：互斥锁保证了某一时刻只能有一个线程占有该资源。

  • 读写锁：将代码功能分为读和写，读不互斥，写互斥。

  • 公平锁/非公平锁：公平锁就是在等待队列里排最前面的的先获得锁，非公平锁就是谁抢到谁用。

  • 重量级锁/轻量级锁/偏向锁：使用操作系统 “Mutex Lock” 功能来实现锁机制的叫重量级锁，因为这种锁成本高；轻量级锁是对重量级锁的优化，提高性能；偏向锁是对轻量级锁的优化，在无多线程竞争的情况下尽量减少不必要的轻量级锁执行路径。

synchronized

属于阻塞锁、互斥锁、非公平锁以及可重入锁，在 JDK1.6 以前属于重量级锁，后来做了优化。

用法：

  • 指定加锁对象

  • 用于静态代码块/方法

  • 用于动态代码块/方法

示例：

  1. public static synchronized void test1(){

  2. System.out.println("test");

  3. }

  4. public synchronized void test2(){

  5. System.out.println("test");

  6. }

  7. public void test3(){

  8. synchronized (Main.class){

  9. System.out.println("test");

  10. }

  11. }

当锁加在静态代码块上或者静态方法上或者为 synchronized(xxx.class){} 时，锁作用于整个类，凡是属于这个类的对象的相关都会被上锁，当用于动态方法或者为或者为synchronized (object){}时锁作用于对象；除此之外，synchronized可以保证线程的可见性和有序性。

Lock

Lock 是一个接口，其下有多个实现类。

方法说明：

  • lock()方法是平常使用得最多的一个方法，就是用来获取锁。如果锁已被其他线程获取，则进行等待。

  • tryLock()方法是有返回值的，它表示用来尝试获取锁，如果获取成功，则返回true，如果获取失败（即锁已被其他线程获取），则返回false，这个方法还可以设置一个获取锁的等待时长，如果时间内获取不到直接返回。

  • 两个线程同时通过lock.lockInterruptibly()想获取某个锁时，假若此时线程A获取到了锁，而线程B只有在等待，那么对线程B调用threadB.interrupt()方法能够中断线程B的等待过程。

  • unLock()方法是用来释放锁。

  • newCondition()：生成一个和线程绑定的Condition实例，利用该实例我们可以让线程在合适的时候等待，在特定的时候继续执行，相当于得到这个线程的wait和notify方法。

ReentrantLock

ReentrantLock 重入锁，是实现 Lock 接口的一个类，它对公平锁和非公平锁都支持，在构造方法中传入一个 boolean 值，true 时为公平锁，false 时为非公平锁。

Semaphore(信号量)

信号量是对锁的扩展，锁每次只允许一个线程访问一个资源，而信号量却可以指定多个线程访问某个资源，信号量的构造函数为

  1. public Semaphore(int permits) {

  2. sync = new NonfairSync(permits);

  3. }

  4. public Semaphore(int permits, boolean fair) {

  5. sync = fair ? new FairSync(permits) : new NonfairSync(permits);

  6. }

第一个方法指定了可使用的线程数，第二个方法的布尔值表示是否为公平锁。

acquire() 方法尝试获得一个许可，如果获取不到则等待；tryAcquire() 方法尝试获取一个许可，成功返回 true，失败返回false，不会阻塞，tryAcquire(int i) 指定等待时间；release() 方法释放一个许可。

ReadWriteLock

读写分离锁， 读写分离锁可以有效的减少锁竞争，读锁是共享锁，可以被多个线程同时获取，写锁是互斥只能被一个线程占有，ReadWriteLock 是一个接口，其中 readLock() 获得读锁，writeLock() 获得写锁 其实现类 ReentrantReadWriteLock 是一个可重入得的读写锁，它支持锁的降级(在获得写锁的情况下可以再持有读锁)，不支持锁的升级（在获得读锁的情况下不能再获得写锁）；读锁和写锁也是互斥的，也就是一个资源要么被上了一个写锁，要么被上了多个读锁，不会发生这个资即被上写锁又被上读锁的情况。

cas

cas(比较替换)：无锁策略的一种实现方式，过程为获取到变量旧值（每个线程都有一份变量值的副本），和变量目前的新值做比较，如果一样证明变量没被其他线程修改过，这个线程就可以更新这个变量，否则不能更新；通俗的说就是通过不加锁的方式来修改共享资源并同时保证安全性。

使用cas的话对于属性变量不能再用传统的 int ,long 等；要使用原子类代替原先的数据类型操作，比如 AtomicBoolean，AtomicInteger，AtomicInteger 等。

并发下集合类

并发集合类主要有：

  • ConcurrentHashMap：支持多线程的分段哈希表，它通过将整个哈希表分成多段的方式减小锁粒度。

  • ConcurrentSkipListMap：ConcurrentSkipListMap的底层是通过跳表来实现的。跳表是一个链表，但是通过使用“跳跃式”查找的方式使得插入、读取数据时复杂度变成了O（logn）。

  • ConCurrentSkipListSet：参考 ConcurrentSkipListMap。

  • CopyOnWriteArrayList：是 ArrayList 的一个线程安全的变形，其中所有可变操作（添加、设置，等等）都是通过对基础数组进行一次新的复制来实现的。

  • CopyOnWriteArraySet：参考 CopyOnWriteArrayList。

  • ConcurrentLinkedQueue：cas 实现的非阻塞并发队列。

线程池

介绍

多线程的设计优点是能很大限度的发挥多核处理器的计算能力，但是，若不控制好线程资源反而会拖累cpu，降低系统性能，这就涉及到了线程的回收复用等一系列问题；而且本身线程的创建和销毁也很耗费资源，因此找到一个合适的方法来提高线程的复用就很必要了。

线程池就是解决这类问题的一个很好的方法：线程池中本身有很多个线程，当需要使用线程的时候拿一个线程出来，当用完则还回去，而不是每次都创建和销毁。在 JDK 中提供了一套 Executor 线程池框架，帮助开发人员有效的进行线程控制。

Executor 使用

获得线程池的方法：

  • newFixedThreadPool(int nThreads) ：创建固定数目线程的线程池。

  • newCachedThreadPool：创建一个可缓存的线程池，调用execute将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线 程并添加到池中。

  • newSingleThreadExecutor：创建一个单线程化的 Executor。

  • newScheduledThreadPool：创建一个支持定时及周期性的任务执行的线程池。

以上方法都是返回一个 ExecutorService 对象，executorService.execute() 传入一个 Runnable 对象，可执行一个线程任务。

下面看示例代码

  1. public class Test implements Runnable{

  2. int i=0;

  3. public Test(int i){

  4. this.i=i;

  5. }

  6. public void run() {

  7. System.out.println(Thread.currentThread().getName()+"====="+i);

  8. }

  9. public static void main(String[] args) throws InterruptedException {

  10. ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

  11. for(int i=0;i<10;i++){

  12. cachedThreadPool.execute(new Test(i));

  13. Thread.sleep(1000);

  14. }

  15. }

  16. }

线程池是一个庞大而复杂的体系，本文定位是基础，不对其做更深入的研究，感兴趣的小伙伴可以自行查资料进行学习。

ScheduledExecutorService

newScheduledThreadPool(int corePoolSize) 会返回一个ScheduledExecutorService 对象，可以根据时间对线程进行调度；其下有三个执行线程任务的方法：schedule()，scheduleAtFixedRate() 以及 scheduleWithFixedDelay() 该线程池可解决定时任务的问题。

示例：

  1. class Test implements Runnable {

  2. private String testStr;

  3. Test(String testStr) {

  4. this.testStr = testStr;

  5. }

  6. @Override

  7. public void run() {

  8. System.out.println(testStr + " >>>> print");

  9. }

  10. public static void main(String[] args) {

  11. ScheduledExecutorService service = Executors.newScheduledThreadPool(10);

  12. long wait = 1;

  13. long period = 1;

  14. service.scheduleAtFixedRate(new MyScheduledExecutor("job1"), wait, period, TimeUnit.SECONDS);

  15. service.scheduleWithFixedDelay(new MyScheduledExecutor("job2"), wait, period, TimeUnit.SECONDS);

  16. scheduledExecutorService.schedule(new MyScheduledExecutor("job3"), wait, TimeUnit.SECONDS);//延时waits 执行

  17. }

  18. }

job1的执行方式是任务发起后间隔 wait 秒开始执行，每隔 period 秒(注意：不包括上一个线程的执行时间)执行一次；

job2的执行方式是任务发起后间隔 wait 秒开始执行，等线程结束后隔 period 秒开始执行下一个线程；

job3只执行一次，延迟 wait 秒执行；

ScheduledExecutorService 还可以配合 Callable 使用来回调获得线程执行结果，还可以取消队列中的执行任务等操作，这属于比较复杂的用法，我们这里掌握基本的即可，到实际遇到相应的问题时我们在现学现用，节省学习成本。

锁优化

减小锁持有时间

减小锁的持有时间可有效的减少锁的竞争。如果线程持有锁的时间越长，那么锁的竞争程度就会越激烈。因此，应尽可能减少线程对某个锁的占有时间，进而减少线程间互斥的可能。

减少锁持有时间的方法有：

  • 进行条件判断，只对必要的情况进行加锁，而不是整个方法加锁。

  • 减少加锁代码的行数，只对必要的步骤加锁。

减小锁粒度

减小锁的范围，减少锁住的代码行数可减少锁范围，减小共享资源的范围也可减小锁的范围。减小锁共享资源的范围的方式比较常见的有分段锁，比如 ConcurrentHashMap ，它将数据分为了多段，当需要 put 元素的时候，并不是对整个 hashmap 进行加锁，而是先通过 hashcode 来知道他要放在那一个分段中，然后对这个分段进行加锁，所以当多线程 put 的时候，只要不是放在一个分段中，就实现了真正的并行的插入。

锁分离

锁分离最常见的操作就是读写分离了，读写分离的操作参考 ReadWriteLock 章节，而对读写分离进一步的延伸就是锁分离了。为了提高线程的并行量，我们可以针对不同的功能采用不同的锁，而不是统统用同一把锁。比如说有一个同步方法未进行锁分离之前，它只有一把锁，任何线程来了，只有拿到锁才有资格运行，进行锁分离之后就不是这种情形了——来一个线程，先判断一下它要干嘛，然后发一个对应的锁给它，这样就能一定程度上提高线程的并行数。

锁粗化

一般为了保证多线程间的有效并发，会要求每个线程持有锁的时间尽量短，也就是说锁住的代码尽量少。但是如果如果对同一个锁不停的进行请求、同步和释放，其本身也会消耗系统宝贵的资源，反而不利于性能的优化 。比如有三个步骤：a、b、c，a同步，b不同步，c同步；那么一个线程来时候会上锁释放锁然后又上锁释放锁。这样反而可能会降低线程的执行效率，这个时候我们将锁粗化可能会更好——执行 a 的时候上锁，执行完 c 再释放锁。

锁扩展

分布式锁

JDK 提供的锁在单体项目中不会有什么问题，但是在集群项目中就会有问题了。在分布式模型下，数据只有一份（或有限制），此时需要利用锁的技术控制某一时刻修改数据的进程数。JDK 锁显然无法满足我们的需求，于是就有了分布式锁。

分布式锁的实现有三种方式：

  • 基于数据库实现分布式锁

  • 基于缓存（redis，memcached，tair）实现分布式锁

  • 基于 Zookeeper 实现分布式锁

基于redis的分布式锁比较使用普遍，在这里介绍其原理和使用：

redis 实现锁的机制是 setnx 指令，setnx 是原子操作命令，锁存在不能设置值，返回 0 ；锁不存在，则设置锁，返回 1 ，根据返回值来判断上锁是否成功。看到这里你可能想为啥不先 get 有没有值，再 set 上锁；首先我们要知道，redis 是单线程的，同一时刻只可能有一个线程操作内存，然后 setnx 是一个操作步骤(具有原子性)，而 get 再 set 是两个步骤（不具有原子性）。如果使用第二种可能会发生这种情况：客户端 a get发现没有锁，这个时候被切换到客户端b，b get也发现没锁，然后b set，这个时候又切换到a客户端 a set；这种情况下，锁完全没起作用。所以，redis分布式锁，原子性是关键。

对于 web 应用中 redis 客户端用的比较多的是 lettuce，jedis，redisson。springboot 的 redis 的 start 包底层是 lettuce ，但对 redis 分布式锁支持得最好的是 redisson（如果用 redisson 你就享受不到 redis 自动化配置的好处了）；不过 springboot 的 redisTemplete 支持手写 lua 脚本，我们可以通过手写 lua 脚本来实现 redis 锁。

代码示例：

  1. public boolean lockByLua(String key, String value, Long expiredTime){

  2. String strExprie = String.valueOf(expiredTime);

  3. StringBuilder sb = new StringBuilder();

  4. sb.append("if redis.call(\"setnx\",KEYS[1],ARGV[1])==1 ");

  5. sb.append("then ");

  6. sb.append(" redis.call(\"pexpire\",KEYS[1],KEYS[2]) ");

  7. sb.append(" return 1 ");

  8. sb.append("else ");

  9. sb.append(" return 0 ");

  10. sb.append("end ");

  11. String script = sb.toString();

  12. RedisCallback<Boolean> callback = (connection) -> {

  13. return connection.eval(script.getBytes(), ReturnType.BOOLEAN, 2, key.getBytes(Charset.forName("UTF-8")),strExprie.getBytes(Charset.forName("UTF-8")), value.getBytes(Charset.forName("UTF-8")));

  14. };

  15. Boolean execute = stringRedisTemplate.execute(callback);

  16. return execute;

  17. }

关于lua脚本的语法我就不做介绍了。

在 github 上也有开源的 redis 锁项目，比如 spring-boot-klock-starter 感兴趣的小伙伴可以去试用一下。

数据库锁

对于存在多线程问题的项目，比如商品货物的进销存，订单系统单据流转这种，我们可以通过代码上锁来控制并发，也可以使用数据库锁来控制并发，数据库锁从机制上来说分乐观锁和悲观锁。

悲观锁：

悲观锁分为共享锁（S锁）和排他锁（X锁），MySQL 数据库读操作分为三种——快照读，当前读；快照读就是普通的读操作，如：

  1. select *from table

当前读就是对数据库上悲观锁了；其中 select...lockinshare mode 属于共享锁，多个事务对于同一数据可以共享，但只能读不能修改。而下面三种 SQL :

  1. select ...for update

  2. update ... set...

  3. insert into ...

属于排他锁，排他锁就是不能与其他锁并存，如一个事务获取了一个数据行的排他锁，其他事务就不能再获取该行的其他锁，包括共享锁和排他锁，但是获取排他锁的事务是可以对数据行读取和修改，排他锁是阻塞锁。

乐观锁：

就是很乐观，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，如果有则更新失败。一种实现方式为在数据库表中加一个版本号字段 version ，任何 update 语句 where 后面都要跟上 version=？，并且每次 update 版本号都加 1。如果 a 线程要修改某条数据，它需要先 select 快照读获得版本号，然后 update ，同时版本号加一。这样就保证了在 a 线程修改某条数据的时候，确保其他线程没有修改过这条数据，一旦其他线程修改过，就会导致 a 线程版本号对不上而更新失败（这其实是一个简化版的mvcc）。

乐观锁适用于允许更新失败的业务场景，悲观锁适用于确保更新操作被执行的场景。

并发编程相关

  • 善用 Java8 Stream

  • 对于生产者消费者模式，条件判断是使用 while 而不是 if

  • 懒汉单例采用双重检查和锁保证线程安全

  • 善用 Future 模式

  • 合理使用 ThreadLocal

Java 8 引入 lambda 表达式使在 Java 中使用函数式编程很方便。而 Java 8 中的 stream 对数据的处理能使线程执行速度得以优化。Future 模式是一种对异步线程的回调机制；现在 cpu 都是多核的，我们在处理一些较为费时的任务时可使用异步，在后台开启多个线程同时处理，等到异步线程处理完再通过 Future 回调拿到处理的结果。

ThreadLocal 的实例代表了一个线程局部的变量，每条线程都只能看到自己的值，并不会意识到其它的线程中也存在该变量(这里原理就不说了，网上资料很多)，总之就是我们如果想在多线程的类里面使用线程安全的变量就用 ThreadLocal ，但是请一定要注意用完记得 remove ，不然会发生内存泄漏。