【高并发】AQS案例详解

举报
冰 河 发表于 2023/02/28 12:45:04 2023/02/28
【摘要】 AQS内部维护了一个CLH队列来管理锁,线程会首先尝试获取锁,如果失败,就将当前线程以及等待等信息封装成一个Node节点,加入到同步队列SyncQueue,接着会不断循环尝试获取锁,获取锁的条件是当前节点为Head的直接后继节点才会尝试获取锁,如果失败,就会阻塞自己,直到自己被唤醒。而持有锁的线程释放锁的时候,会唤醒队列中的后继线程。基于这些基础的设计和思路,JDK提供了许多基于AQS的子类

AQS的全称为AbstractQueuedSynchronizer,是在J.U.C(java.util.concurrent)下子包中的类。

一、AQS的设计如下

(1)使用Node实现FIFO队列,可以用于构建锁或者其他同步装置的基础框架。
(2)利用了一个int类型表示状态
在AQS类中,有一个叫做state的成员变量。
基于AQS有一个同步组件ReentrantLock,在ReentrantLock中,state表示获取锁的线程数。如果state=0,则表示还没有线程获取锁;如果state=1,则表示有线程获取了锁;如果state>1,则表示重入锁的数量。
(3)使用方法是继承
设计上基于模板方法,使用时需要继承AQS,并覆写其中的方法
(4)子类通过继承并通过实现它的方法管理其状态{acquire和release}的方法操纵状态
(5)可以同时实现排它锁和共享锁模式(独占、共享)
站在使用者的角度,AQS的功能主要分为两类:独占模式和共享模式。它的所有子类中要么实现并使用了它的独占功能的API,要么使用了共享锁的功能,而不会同时使用两套API。即便是它最有名的子类——ReentrantReadWriteLock,也是通过两个内部类——ReadLock(读锁)和WriteLock(写锁)两套API来实现的。

二、AQS内部实现的大体思路

首先,AQS内部维护了一个CLH队列来管理锁,线程会首先尝试获取锁,如果失败,就将当前线程以及等待等信息封装成一个Node节点,加入到同步队列SyncQueue,接着会不断循环尝试获取锁,获取锁的条件是当前节点为Head的直接后继节点才会尝试获取锁,如果失败,就会阻塞自己,直到自己被唤醒。而持有锁的线程释放锁的时候,会唤醒队列中的后继线程。基于这些基础的设计和思路,JDK提供了许多基于AQS的子类,比如:CountDownLatch、Semaphore、CyclicBarrier、ReentrantLock、Condition、FutureTask等

三、AQS同步组件

  • CountDownLatch:闭锁,通过一个计数,来保证线程是否一直阻塞
  • Semaphore:控制同一时间并发线程的数目
  • CyclicBarrier:与CountDownLatch类似,都能阻塞进程;
  • ReentrantLock:可重入锁
  • Condition: 在使用时需要ReentrantLock
  • FutureTask:对比Runnable和Callable

1.CountDownLatch

同步辅助类,通过它可以阻塞当前线程。也就是说,能够实现一个线程或者多个线程一直等待,直到其他线程执行的操作完成。使用一个给定的计数器进行初始化,该计数器的操作是原子操作,即同时只能有一个线程操作该计数器。
调用该类await()方法的线程会一直阻塞,直到其他线程调用该类的countDown()方法,使当前计数器的值变为0为止。每次调用该类的countDown()方法,当前计数器的值就会减1。当计数器的值减为0的时候,所有因调用await()方法而处于等待状态的线程就会继续往下执行。这种操作只能出现一次,因为该类中的计数器不能被重置。如果需要一个可以重置计数次数的版本,可以考虑使用CyclicBarrier类。
CountDownLatch支持给定时间的等待,超过一定的时间不再等待,使用时只需要在await()方法中传入需要等待的时间即可。此时,await()方法的方法签名如下:

public boolean await(long timeout, TimeUnit unit)

CountDownLatch使用场景
在某些业务场景中,程序执行需要等待某个条件完成后才能继续执行后续的操作。典型的应用为并行计算:当某个处理的运算量很大时,可以将该运算任务拆分成多个子任务,等待所有的子任务都完成之后,父任务再拿到所有子任务的运算结果进行汇总。

调用ExecutorService类的shutdown()方法,并不会第一时间内把所有线程全部都销毁掉,而是让当前已有的线程全部执行完,之后,再把线程池销毁掉。

示例代码如下:

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CountDownLatchExample {
    private static final int threadCount = 200;

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

        ExecutorService exec = Executors.newCachedThreadPool();
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        log.info("finish");
        exec.shutdown();
    }

    private static void test(int threadNum) throws InterruptedException {
        Thread.sleep(100);
        log.info("{}", threadNum);
        Thread.sleep(100);
    }
}

支持给定时间等待的示例代码如下:

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@Slf4j
public class CountDownLatchExample {
    private static final int threadCount = 200;

    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newCachedThreadPool();
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await(10, TimeUnit.MICROSECONDS);
        log.info("finish");
        exec.shutdown();
    }

    private static void test(int threadNum) throws InterruptedException {
        Thread.sleep(100);
        log.info("{}", threadNum);
    }
}

2.Semaphore

控制同一时间并发线程的数目。能够完成对于信号量的控制,可以控制某个资源可被同时访问的个数。
提供了两个核心方法——acquire()方法和release()方法。acquire()方法表示获取一个许可,如果没有则等待,release()方法则是在操作完成后释放对应的许可。Semaphore维护了当前访问的个数,通过提供同步机制来控制同时访问的个数。Semaphore可以实现有限大小的链表。
Semaphore使用场景如
Semaphore常用于仅能提供有限访问的资源,比如:数据库连接数

每次获取并释放一个许可,示例代码如下:

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
    private static final int threadCount = 200;

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

        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore  = new Semaphore(3);

        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    semaphore.acquire();  //获取一个许可
                    test(threadNum);
                    semaphore.release();  //释放一个许可
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        exec.shutdown();
    }

    private static void test(int threadNum) throws InterruptedException {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

每次获取并释放多个许可,示例代码如下:

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
    private static final int threadCount = 200;

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

        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore  = new Semaphore(3);

        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    semaphore.acquire(3);  //获取多个许可
                    test(threadNum);
                    semaphore.release(3);  //释放多个许可
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        log.info("finish");
        exec.shutdown();
    }

    private static void test(int threadNum) throws InterruptedException {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

假设有这样一个场景,并发太高了,即使使用Semaphore进行控制,处理起来也比较棘手。假设系统当前允许的最高并发数是3,超过3后就需要丢弃,使用Semaphore也能实现这样的场景,示例代码如下:

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@Slf4j
public class SemaphoreExample {
    private static final int threadCount = 200;

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

        ExecutorService exec = Executors.newCachedThreadPool();
        final Semaphore semaphore  = new Semaphore(3);

        for (int i = 0; i < threadCount; i++){
            final int threadNum = i;
            exec.execute(() -> {
                try {
	            //尝试获取一个许可,也可以尝试获取多个许可,
                    //支持尝试获取许可超时设置,超时后不再等待后续线程的执行
                    //具体可以参见Semaphore的源码
                    if (semaphore.tryAcquire()) { 
                        test(threadNum);
                        semaphore.release();  //释放一个许可
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        log.info("finish");
        exec.shutdown();
    }
    private static void test(int threadNum) throws InterruptedException {
        log.info("{}", threadNum);
        Thread.sleep(1000);
    }
}

3.CyclicBarrier

是一个同步辅助类,允许一组线程相互等待,直到到达某个公共的屏障点,通过它可以完成多个线程之间相互等待,只有当每个线程都准备就绪后,才能各自继续往下执行后面的操作。
与CountDownLatch有相似的地方,都是使用计数器实现,当某个线程调用了CyclicBarrier的await()方法后,该线程就进入了等待状态,而且计数器执行加1操作,当计数器的值达到了设置的初始值,调用await()方法进入等待状态的线程会被唤醒,继续执行各自后续的操作。CyclicBarrier在释放等待线程后可以重用,所以,CyclicBarrier又被称为循环屏障。

CyclicBarrier使用场景
可以用于多线程计算数据,最后合并计算结果的场景

CyclicBarrier与CountDownLatch的区别
(1)CountDownLatch的计数器只能使用一次,而CyclicBarrier的计数器可以使用reset()方法进行重置,并且可以循环使用
(2)CountDownLatch主要实现1个或n个线程需要等待其他线程完成某项操作之后,才能继续往下执行,描述的是1个或n个线程等待其他线程的关系。而CyclicBarrier主要实现了多个线程之间相互等待,直到所有的线程都满足了条件之后,才能继续执行后续的操作,描述的是各个线程内部相互等待的关系。
(3)CyclicBarrier能够处理更复杂的场景,如果计算发生错误,可以重置计数器让线程重新执行一次。
CyclicBarrier中提供了很多有用的方法,比如:可以通过getNumberWaiting()方法获取阻塞的线程数量,通过isBroken()方法判断阻塞的线程是否被中断。

示例代码如下:

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample {

    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);

    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++){
            final int threadNum = i;
            Thread.sleep(1000);
            executorService.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
		executorService.shutdown();
    }
    private static void race(int threadNum) throws Exception{
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        cyclicBarrier.await();
        log.info("{} continue", threadNum);
    }
}

设置等待超时示例代码如下:

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;
@Slf4j
public class CyclicBarrierExample {

    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5);

    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++){
            final int threadNum = i;
            Thread.sleep(1000);
            executorService.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
    private static void race(int threadNum) throws Exception{
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        try{
            cyclicBarrier.await(2000, TimeUnit.MILLISECONDS);
        }catch (BrokenBarrierException | TimeoutException e){
            log.warn("BarrierException", e);
        }
        log.info("{} continue", threadNum);
    }
}

在声明CyclicBarrier的时候,还可以指定一个Runnable,当线程达到屏障的时候,可以优先执行Runnable中的方法。
示例代码如下:

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class CyclicBarrierExample {

    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> {
        log.info("callback is running");
    });

    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++){
            final int threadNum = i;
            Thread.sleep(1000);
            executorService.execute(() -> {
                try {
                    race(threadNum);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        executorService.shutdown();
    }
    private static void race(int threadNum) throws Exception{
        Thread.sleep(1000);
        log.info("{} is ready", threadNum);
        cyclicBarrier.await();
        log.info("{} continue", threadNum);
    }
}

4.ReentrantLock与锁

Java中主要分为两类锁,一类是synchronized修饰的锁,另外一类就是J.U.C中提供的锁。J.U.C中提供的核心锁就是ReentrantLock。

ReentrantLock(可重入锁)与synchronized区别:
(1)可重入性
二者都是同一个线程进入1次,锁的计数器就自增1,需要等到锁的计数器下降为0时,才能释放锁。
(2)锁的实现
synchronized是基于JVM实现的,而ReentrantLock是JDK实现的
(3)性能的区别
synchronized优化之前性能比ReentrantLock差很多,但是自从synchronized引入了偏向锁,轻量级锁也就是自旋锁后,性能就差不多了。
(4)功能区别

  • 便利性:synchronized使用起来比较方便,并且由编译器保证加锁和释放锁;ReentrantLock需要手工声明加锁和释放锁,最好是在finally代码块中声明释放锁。
  • 锁的灵活度和细粒度:在这点上ReentrantLock会优于synchronized

ReentrantLock独有的功能如下:
(1)ReentrantLock可指定是公平锁还是非公平锁。而synchronized只能是非公平锁。所谓的公平锁就是先等待的线程先获得锁。
(2)提供了一个Condition类,可以分组唤醒需要唤醒的线程。而synchronized只能随机唤醒一个线程,或者唤醒全部的线程
(3)提供能够中断等待锁的线程的机制,lock.lockInterruptibly()。ReentrantLock实现是一种自旋锁,通过循环调用CAS操作来实现加锁,性能上比较好是因为避免了使线程进入内核态的阻塞状态。

synchronized能做的事情ReentrantLock都能做,而ReentrantLock有些能做的事情,synchronized不能做。
在性能上,ReentrantLock不会比synchronized差。

synchronized的优势:
(1)不用手动释放锁,JVM自动处理,如果出现异常,JVM也会自动释放锁
(2)JVM用synchronized进行管理锁定请求和释放时,JVM在生成线程转储时能够锁定信息,这些对调试非常有价值,因为它们能标识死锁或者其他异常行为的来源。而ReentrantLock只是普通的类,JVM不知道具体哪个线程拥有lock对象。
(3)synchronized可以在所有JVM版本中工作,ReentrantLock在某些1.5之前版本的JVM中可能不支持

ReentrantLock中的部分方法说明:

  • boolean tryLock():仅在调用时锁定未被另一个线程保持的情况下才获取锁定
  • boolean tryLock(long, TimeUnit): 如果锁定在给定的等待时间内没有被另一个线程保持,且当前线程没有被中断,则获取这个锁定。
  • void lockInterruptibly():如果当前线程没有被中断,就获取锁定;如果被中断,就抛出异常
  • boolean isLocked():查询此锁定是否由任意线程保持
  • boolean isHeldByCurrentThread(): 查询当前线程是否保持锁定状态;
  • boolean isFair():判断是否是公平锁
  • boolean hasQueuedThread(Thread):查询指定线程是否在等待获取此锁定
  • boolean hasQueuedThreads():查询是否有线程正在等待获取此锁定
  • boolean getHoldCount():查询当前线程保持锁定的个数

示例代码如下:

package io.binghe.concurrency.example.lock;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class LockExample {
    //请求总数
    public static int clientTotal = 5000;
    //同时并发执行的线程数
    public static int threadTotal = 200;
    public static int count = 0;
    private static final Lock lock = new ReentrantLock();
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for(int i = 0; i < clientTotal; i++){
            executorService.execute(() -> {
                try{
                    semaphore.acquire();
                    add();
                    semaphore.release();
                }catch (Exception e){
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("count:{}", count);
    }
    private static void add(){
        lock.lock();
        try{
            count ++;
        }finally {
            lock.unlock();
        }
    }
}

5.ReentrantReadWriteLock

在没有任何读写锁的时候,才可以取得写锁。如果一直有读锁存在,则无法执行写锁,这就会导致写锁饥饿。
示例代码如下:

package io.binghe.concurrency.example.lock;

import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@Slf4j
public class LockExample {

    private final Map<String, Data> map = new TreeMap<>();
    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private final Lock readLock = lock.readLock();
    private final Lock writeLock = lock.writeLock();

    public Data get(String key){
        readLock.lock();
        try{
            return map.get(key);
        }finally {
            readLock.unlock();
        }
    }

    public Set<String> getAllKeys(){
        readLock.lock();
        try{
            return map.keySet();
        }finally {
            readLock.unlock();
        }
    }

    public Data put(String key, Data value){
        writeLock.lock();
        try{
            return map.put(key, value);
        }finally {
            writeLock.unlock();
        }
    }

    class Data{

    }
}

6.StampedLock

控制锁三种模式:写、读、乐观读。
StampedLock的状态由版本和模式两个部分组成,锁获取方法返回的是一个数字作为票据,用相应的锁状态来表示并控制相关的访问,数字0表示没有写锁被授权访问。
在读锁上分为悲观锁和乐观锁,乐观读就是在读操作很多,写操作很少的情况下,可以乐观的认为写入和读取同时发生的几率很小。因此,不悲观的使用完全的读取锁定。程序可以查看读取资料之后,是否遭到写入进行了变更,再采取后续的措施,这样的改进可以大幅度提升程序的吞吐量。
总之,在读线程越来越多的场景下,StampedLock大幅度提升了程序的吞吐量。

StampedLock源码中的案例如下,这里加上了注释

class Point {
	private double x, y;
	private final StampedLock sl = new StampedLock();

	void move(double deltaX, double deltaY) { // an exclusively locked method
		long stamp = sl.writeLock();
		try {
			x += deltaX;
			y += deltaY;
		} finally {
			sl.unlockWrite(stamp);
		}
	}

	//下面看看乐观读锁案例
	double distanceFromOrigin() { // A read-only method
		long stamp = sl.tryOptimisticRead(); //获得一个乐观读锁
		double currentX = x, currentY = y;  //将两个字段读入本地局部变量
		if (!sl.validate(stamp)) { //检查发出乐观读锁后同时是否有其他写锁发生?
			stamp = sl.readLock();  //如果没有,我们再次获得一个读悲观锁
			try {
				currentX = x; // 将两个字段读入本地局部变量
				currentY = y; // 将两个字段读入本地局部变量
			} finally {
				sl.unlockRead(stamp);
			}
		}
		return Math.sqrt(currentX * currentX + currentY * currentY);
	}

	//下面是悲观读锁案例
	void moveIfAtOrigin(double newX, double newY) { // upgrade
		// Could instead start with optimistic, not read mode
		long stamp = sl.readLock();
		try {
			while (x == 0.0 && y == 0.0) { //循环,检查当前状态是否符合
				long ws = sl.tryConvertToWriteLock(stamp); //将读锁转为写锁
				if (ws != 0L) { //这是确认转为写锁是否成功
					stamp = ws; //如果成功 替换票据
					x = newX; //进行状态改变
					y = newY;  //进行状态改变
					break;
				} else { //如果不能成功转换为写锁
					sl.unlockRead(stamp);  //我们显式释放读锁
					stamp = sl.writeLock();  //显式直接进行写锁 然后再通过循环再试
				}
			}
		} finally {
			sl.unlock(stamp); //释放读锁或写锁
		}
	}
}

示例代码如下:

package io.binghe.concurrency.example.lock;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.StampedLock;
@Slf4j
public class LockExample {
    //请求总数
    public static int clientTotal = 5000;
    //同时并发执行的线程数
    public static int threadTotal = 200;

    public static int count = 0;

    private static final StampedLock lock = new StampedLock();

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for(int i = 0; i < clientTotal; i++){
            executorService.execute(() -> {
                try{
                    semaphore.acquire();
                    add();
                    semaphore.release();
                }catch (Exception e){
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("count:{}", count);
    }

    private static void add(){
	//加锁时返回一个long类型的票据
        long stamp = lock.writeLock();
        try{
            count ++;
        }finally {
	    //释放锁的时候带上加锁时返回的票据
            lock.unlock(stamp);
        }
    }
}

总结:
(1)当只有少量竞争者时,synchronized是一个很好的通用锁实现
(2)竞争者不少,但是线程的增长趋势是可预估的,此时,ReentrantLock是一个很好的通用锁实现
(3)synchronized不会引发死锁,其他的锁使用不当可能会引发死锁。

7.Condition

Condition是一个多线程间协调通信的工具类,Condition除了实现wait和notify的功能以外,它的好处在于一个lock可以创建多个Condition,可以选择性的通知wait的线程
特点:
(1)Condition 的前提是Lock,由AQS中newCondition()方法 创建Condition的对象
(2)Condition await方法表示线程从AQS中移除,并释放线程获取的锁,并进入Condition等待队列中等待,等待被signal
(3)Condition signal方法表示唤醒对应Condition等待队列中的线程节点,并加入AQS中,准备去获取锁。

示例代码如下:

package io.binghe.concurrency.example.lock;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public class LockExample {
    public static void main(String[] args) {
        ReentrantLock reentrantLock = new ReentrantLock();
        Condition condition = reentrantLock.newCondition();

        new Thread(() -> {
            try {
                reentrantLock.lock();
                log.info("wait signal"); // 1
                condition.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("get signal"); // 4
            reentrantLock.unlock();
        }).start();

        new Thread(() -> {
            reentrantLock.lock();
            log.info("get lock"); // 2
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            condition.signalAll();
            log.info("send signal ~ "); // 3
            reentrantLock.unlock();
        }).start();
    }
}
【版权声明】本文为华为云社区用户原创内容,转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息, 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。