详解AQS的7个同步组件-低调大师

详解AQS的7个同步组件

2023-03-13 200

摘要：AQS的全称为Abstract Queued Synchronizer，是在J.U.C（java.util.concurrent）下子包中的类。

本文分享自华为云社区《【高并发】AQS案例详解》，作者：冰河。

AQS的全称为Abstract Queued Synchronizer，是在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();
 }
}

点击关注，第一时间了解华为云新鲜技术~

微信关注我们

原文链接：https://my.oschina.net/u/4526289/blog/8563718

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

时效准确率提升之承运商路由网络挖掘｜得物技术

1引子履约时长是电商的生命线，直接关系到用户的消费体验。新华网[5]2022年双十一的报告显示，37.4%的受访者希望次日达，29.91%希望当日达。相较于其他物品，受访者对手机、电脑、数码产品的物流时效要求更高，更希望当日或1-2天内能收到货。得物履约场景中，主要的阶段包括仓库内生产和第三方承运商配送。在用户支付时，得物会根据仓库的生产情况和运配资源，给用户一个承诺时效。 1.1 为什么要预测承运商的线路时效在履约过程中，得物需要监控订单的流转，及时的发现可能超时的订单（与和用户承诺时效相比），这里包含仓库生产的监控和三方配送的监控。在实际过程中我们发现：配送节点发生变更时，承运商给的预测偏保守的。下面例子中，到了营业部承运商才给到比较精准的预计送达时间，故在分拣中心使用承运商的预计送达时间容易出现误报。下图是承运商接口返回的预计送达时效的宽松指数，可以看到在接近目的地时，承诺时效才比较准确。 2承运商网络是如何运作的在构建承运商网络之前，需要先了解承运商网络是如何工作的。下面是从A网点到E网点的配送示意图，分为以下内容：（1）节点，包含的揽收和派送网点以及分拣中心。（...

2023-03-10

416

【前言】了解一个组件的最好方式是先使用该组件，今天我们就来聊聊如何通过java api对iceberg进行操作。为什么是选择api进行介绍，而不是更通用的flink、spark、hive等。一方面是觉得flink、spark使用iceberg的介绍网上已经有很多，官网的介绍也比较清晰，而java api的介绍则相对少些；另一方面，不管是flink，spark最终都还是调用这些基本的api完成相关的操作的，因此先从api入手，后续对flink，spark，trino等组件对iceberg的操作原理理解起来也会更容易些。所以就有了本文的内容。【catalog的创建】在创建数据库，表之前需要先创建catalog，这里主要介绍hive类型的catalog。 importorg.apache.iceberg.hive.HiveCatalog;HiveCatalog catalog = newHiveCatalog();catalog.setConf(conf);Map <String, String> properties = newHashMap<String, St...

2023-03-10

508

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Oracle

Oracle Database，又名Oracle RDBMS，或简称Oracle。是甲骨文公司的一款关系数据库管理系统。它是在数据库领域一直处于领先地位的产品。可以说Oracle数据库系统是目前世界上流行的关系数据库管理系统，系统可移植性好、使用方便、功能强，适用于各类大、中、小、微机环境。它是一种高效率、可靠性好的、适应高吞吐量的数据库方案。

Apache Tomcat

Tomcat是Apache 软件基金会（Apache Software Foundation）的Jakarta 项目中的一个核心项目，由Apache、Sun 和其他一些公司及个人共同开发而成。因为Tomcat 技术先进、性能稳定，而且免费，因而深受Java 爱好者的喜爱并得到了部分软件开发商的认可，成为目前比较流行的Web 应用服务器。

Eclipse

Eclipse 是一个开放源代码的、基于Java的可扩展开发平台。就其本身而言，它只是一个框架和一组服务，用于通过插件组件构建开发环境。幸运的是，Eclipse 附带了一个标准的插件集，包括Java开发工具（Java Development Kit，JDK）。