From 30d826bb8e5b0b43b847c161120494155c3cd539 Mon Sep 17 00:00:00 2001
From: wyl <959814898@qq.com>
Date: Mon, 1 Jul 2024 00:28:32 +0800
Subject: [PATCH] =?UTF-8?q?java=E5=9F=BA=E7=A1=80=20=E5=B9=B6=E5=8F=91?=
=?UTF-8?q?=E6=B5=8B=E8=AF=95=E4=BB=A3=E7=A0=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
{concurrent => javaBase}/pom.xml | 7 +-
.../concurrent/model/BlockingQueueTest.java | 29 ++
.../java/concurrent/model/QueueConsumer.java | 0
.../java/concurrent/model/QueueProducer.java | 0
.../java/concurrent/model/WaitConsumer.java | 0
.../java/concurrent/model/WaitProducer.java | 1 -
.../main/java/concurrent/model/WaitTest.java | 24 +
.../java/concurrent/pool/ThreadPoolTest.java | 110 +++++
.../concurrent/queue/BlockingQueueTest.java | 23 +-
.../java/concurrent/queue/DelayedQueue.java | 74 +++
.../concurrent/reentrantlock/AOSLockTest.java | 99 ++++
.../concurrent/reentrantlock/CLHLock.java | 114 +++++
.../concurrent/reentrantlock/CLHLockTest.java | 27 ++
.../reentrantlock/ReentrantLockTest.java | 133 ++++++
.../java/concurrent/thread/SemaphoreTest.java | 62 +++
.../concurrent/thread/SyncronizedLevelUp.java | 25 +
.../concurrent/thread/SyncronizedTest.java | 75 +++
.../concurrent/ttl/context/ContextUtil.java | 0
.../concurrent/ttl/task/CallableTask.java | 0
.../concurrent/ttl/task/RunnableTask.java | 0
.../completableFutureTest.java | 432 ++++++++++++++++++
.../test/java/concurrent/model/ModelTest.java | 48 ++
.../ttl/TransmittableThreadLocalTest.java | 0
23 files changed, 1258 insertions(+), 25 deletions(-)
rename {concurrent => javaBase}/pom.xml (82%)
create mode 100644 javaBase/src/main/java/concurrent/model/BlockingQueueTest.java
rename {concurrent => javaBase}/src/main/java/concurrent/model/QueueConsumer.java (100%)
rename {concurrent => javaBase}/src/main/java/concurrent/model/QueueProducer.java (100%)
rename {concurrent => javaBase}/src/main/java/concurrent/model/WaitConsumer.java (100%)
rename {concurrent => javaBase}/src/main/java/concurrent/model/WaitProducer.java (96%)
create mode 100644 javaBase/src/main/java/concurrent/model/WaitTest.java
create mode 100644 javaBase/src/main/java/concurrent/pool/ThreadPoolTest.java
rename concurrent/src/main/java/concurrent/queue/BlockingQueueTestWork.java => javaBase/src/main/java/concurrent/queue/BlockingQueueTest.java (71%)
create mode 100644 javaBase/src/main/java/concurrent/queue/DelayedQueue.java
create mode 100644 javaBase/src/main/java/concurrent/reentrantlock/AOSLockTest.java
create mode 100644 javaBase/src/main/java/concurrent/reentrantlock/CLHLock.java
create mode 100644 javaBase/src/main/java/concurrent/reentrantlock/CLHLockTest.java
create mode 100644 javaBase/src/main/java/concurrent/reentrantlock/ReentrantLockTest.java
create mode 100644 javaBase/src/main/java/concurrent/thread/SemaphoreTest.java
create mode 100644 javaBase/src/main/java/concurrent/thread/SyncronizedLevelUp.java
create mode 100644 javaBase/src/main/java/concurrent/thread/SyncronizedTest.java
rename {concurrent => javaBase}/src/main/java/concurrent/ttl/context/ContextUtil.java (100%)
rename {concurrent => javaBase}/src/main/java/concurrent/ttl/task/CallableTask.java (100%)
rename {concurrent => javaBase}/src/main/java/concurrent/ttl/task/RunnableTask.java (100%)
create mode 100644 javaBase/src/test/java/concurrent/completableFuture/completableFutureTest.java
create mode 100644 javaBase/src/test/java/concurrent/model/ModelTest.java
rename {concurrent => javaBase}/src/test/java/concurrent/ttl/TransmittableThreadLocalTest.java (100%)
diff --git a/concurrent/pom.xml b/javaBase/pom.xml
similarity index 82%
rename from concurrent/pom.xml
rename to javaBase/pom.xml
index 79adf6a..e5e709d 100644
--- a/concurrent/pom.xml
+++ b/javaBase/pom.xml
@@ -9,7 +9,7 @@
4.0.0
- concurrent
+ javaBase
8
@@ -20,10 +20,5 @@
com.alibaba
transmittable-thread-local
-
- cn.hutool
- hutool-core
-
-
\ No newline at end of file
diff --git a/javaBase/src/main/java/concurrent/model/BlockingQueueTest.java b/javaBase/src/main/java/concurrent/model/BlockingQueueTest.java
new file mode 100644
index 0000000..2033294
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/model/BlockingQueueTest.java
@@ -0,0 +1,29 @@
+package concurrent.model;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+
+/**
+ * @description: 生产消费者模型测试使用BlockingQueue
+ * @author: wangyl
+ * @date: 2024/6/30
+ */
+public class BlockingQueueTest {
+ public static void main(String[] args) throws InterruptedException {
+ /**
+ * 使用ArrayBlockingQueue 插入和删除数据,只采用了一个lock
+ */
+// BlockingQueue blockingQueue = new ArrayBlockingQueue<>(5000);
+ /**
+ * 使用LinkedBlockingDeque 是在插入和删除分别采用了putLock和takeLock,这样可以降低线程由于线程无法获取到lock而进入WAITING状态的可能性,从而提高了线程并发执行的效率
+ */
+ BlockingQueue blockingQueue = new LinkedBlockingDeque<>(5000);
+ for (int i = 0; i < 1000; i++) {
+ new QueueProducer(blockingQueue, i).start();
+ }
+ for (int i = 0; i < 200000; i++) {
+ new QueueConsumer(blockingQueue, i).start();
+ }
+ Thread.sleep(100000000);
+ }
+}
diff --git a/concurrent/src/main/java/concurrent/model/QueueConsumer.java b/javaBase/src/main/java/concurrent/model/QueueConsumer.java
similarity index 100%
rename from concurrent/src/main/java/concurrent/model/QueueConsumer.java
rename to javaBase/src/main/java/concurrent/model/QueueConsumer.java
diff --git a/concurrent/src/main/java/concurrent/model/QueueProducer.java b/javaBase/src/main/java/concurrent/model/QueueProducer.java
similarity index 100%
rename from concurrent/src/main/java/concurrent/model/QueueProducer.java
rename to javaBase/src/main/java/concurrent/model/QueueProducer.java
diff --git a/concurrent/src/main/java/concurrent/model/WaitConsumer.java b/javaBase/src/main/java/concurrent/model/WaitConsumer.java
similarity index 100%
rename from concurrent/src/main/java/concurrent/model/WaitConsumer.java
rename to javaBase/src/main/java/concurrent/model/WaitConsumer.java
diff --git a/concurrent/src/main/java/concurrent/model/WaitProducer.java b/javaBase/src/main/java/concurrent/model/WaitProducer.java
similarity index 96%
rename from concurrent/src/main/java/concurrent/model/WaitProducer.java
rename to javaBase/src/main/java/concurrent/model/WaitProducer.java
index f28b8d3..8e91b7e 100644
--- a/concurrent/src/main/java/concurrent/model/WaitProducer.java
+++ b/javaBase/src/main/java/concurrent/model/WaitProducer.java
@@ -1,7 +1,6 @@
package concurrent.model;
import lombok.AllArgsConstructor;
-import lombok.RequiredArgsConstructor;
import java.util.Queue;
import java.util.Random;
diff --git a/javaBase/src/main/java/concurrent/model/WaitTest.java b/javaBase/src/main/java/concurrent/model/WaitTest.java
new file mode 100644
index 0000000..34baa60
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/model/WaitTest.java
@@ -0,0 +1,24 @@
+package concurrent.model;
+
+import java.util.LinkedList;
+import java.util.Queue;
+
+/**
+ * @description: 使用wait 实现生产消息模型
+ * @author: wangyl
+ * @date: 2024/6/30
+ */
+public class WaitTest {
+ public static void main(String[] args) throws InterruptedException {
+ Object o = new Object();
+ Queue queue = new LinkedList();
+ Integer maxSize = 100;
+ for (int i = 0; i < 15; i++) {
+ new WaitProducer(maxSize, o, queue).start();
+ }
+ for (int i = 0; i < 2; i++) {
+ new WaitConsumer(o, queue).start();
+ }
+ Thread.sleep(10000000);
+ }
+}
diff --git a/javaBase/src/main/java/concurrent/pool/ThreadPoolTest.java b/javaBase/src/main/java/concurrent/pool/ThreadPoolTest.java
new file mode 100644
index 0000000..5bb764a
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/pool/ThreadPoolTest.java
@@ -0,0 +1,110 @@
+package concurrent.pool;
+
+import java.util.Objects;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @description: 线程池测试
+ * @author: wangyl
+ * @date: 2024/6/30
+ */
+public class ThreadPoolTest {
+ public static void main(String[] args) throws InterruptedException {
+ NamedThreadFactory namedThreadFactory = new NamedThreadFactory("wyl", false);
+ ExecutorService executorService = new ThreadPoolExecutor(10, 20, 100, TimeUnit.SECONDS,
+ new ArrayBlockingQueue<>(10), namedThreadFactory);
+ executorService.execute(() -> System.out.println(123));
+ }
+
+ /**
+ * @description: 有名字的线程工厂
+ * @author: wangyl
+ * @date: 2024/6/30
+ */
+ static public class NamedThreadFactory implements ThreadFactory {
+ private final String prefix;
+ /**
+ * 线程组
+ */
+ private final ThreadGroup group;
+ /**
+ * 线程组
+ */
+ private final AtomicInteger threadNumber = new AtomicInteger(1);
+ /**
+ * 是否守护线程
+ */
+ private final boolean isDaemon;
+ /**
+ * 无法捕获的异常统一处理
+ */
+ private final Thread.UncaughtExceptionHandler handler;
+
+ /**
+ * 构造
+ *
+ * @param prefix 线程名前缀
+ * @param isDaemon 是否守护线程
+ */
+ public NamedThreadFactory(String prefix, boolean isDaemon) {
+ this(prefix, null, isDaemon);
+ }
+
+ /**
+ * 构造
+ *
+ * @param prefix 线程名前缀
+ * @param threadGroup 线程组,可以为null
+ * @param isDaemon 是否守护线程
+ */
+ public NamedThreadFactory(String prefix, ThreadGroup threadGroup, boolean isDaemon) {
+ this(prefix, threadGroup, isDaemon, null);
+ }
+
+ /**
+ * 构造
+ *
+ * @param prefix 线程名前缀
+ * @param threadGroup 线程组,可以为null
+ * @param isDaemon 是否守护线程
+ * @param handler 未捕获异常处理
+ */
+ public NamedThreadFactory(String prefix, ThreadGroup threadGroup, boolean isDaemon, Thread.UncaughtExceptionHandler handler) {
+ this.prefix = Objects.isNull(prefix) ? "wyl" : prefix;
+ if (null == threadGroup) {
+ final SecurityManager s = System.getSecurityManager();
+ threadGroup = (null != s) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
+ }
+ this.group = threadGroup;
+ this.isDaemon = isDaemon;
+ this.handler = handler;
+ }
+
+ @Override
+ public Thread newThread(Runnable r) {
+ final Thread t = new Thread(this.group, r, String.format("{}{}", prefix, threadNumber.getAndIncrement()));
+
+ //守护线程
+ if (false == t.isDaemon()) {
+ if (isDaemon) {
+ // 原线程为非守护则设置为守护
+ t.setDaemon(true);
+ }
+ } else if (false == isDaemon) {
+ // 原线程为守护则还原为非守护
+ t.setDaemon(false);
+ }
+ //异常处理
+ if (null != this.handler) {
+ t.setUncaughtExceptionHandler(handler);
+ }
+ //优先级
+ if (Thread.NORM_PRIORITY != t.getPriority()) {
+ // 标准优先级
+ t.setPriority(Thread.NORM_PRIORITY);
+ }
+ return t;
+ }
+ }
+}
diff --git a/concurrent/src/main/java/concurrent/queue/BlockingQueueTestWork.java b/javaBase/src/main/java/concurrent/queue/BlockingQueueTest.java
similarity index 71%
rename from concurrent/src/main/java/concurrent/queue/BlockingQueueTestWork.java
rename to javaBase/src/main/java/concurrent/queue/BlockingQueueTest.java
index 92f5be4..dc00f4f 100644
--- a/concurrent/src/main/java/concurrent/queue/BlockingQueueTestWork.java
+++ b/javaBase/src/main/java/concurrent/queue/BlockingQueueTest.java
@@ -1,32 +1,15 @@
package concurrent.queue;
import lombok.AllArgsConstructor;
-import lombok.SneakyThrows;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
@AllArgsConstructor
-public class BlockingQueueTestWork extends Thread {
+public class BlockingQueueTest {
ArrayBlockingQueue arrayBlockingQueue;
Integer methodNumber;
- @SneakyThrows
- @Override
- public void run() {
- switch (methodNumber) {
- case 1:
- offerTest();
- break;
- case 2:
- addTest();
- break;
- case 3:
- putTest();
- break;
- }
- }
-
public void offerTest() {
ConcurrentLinkedQueue concurrentLinkedQueue = new ConcurrentLinkedQueue<>();
boolean offer = arrayBlockingQueue.offer(1);
@@ -50,5 +33,9 @@ public class BlockingQueueTestWork extends Thread {
arrayBlockingQueue.put(1);
System.out.println("添加成功");
}
+
+ public static void main(String[] args) {
+
+ }
}
diff --git a/javaBase/src/main/java/concurrent/queue/DelayedQueue.java b/javaBase/src/main/java/concurrent/queue/DelayedQueue.java
new file mode 100644
index 0000000..cfc397a
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/queue/DelayedQueue.java
@@ -0,0 +1,74 @@
+package concurrent.queue;
+
+import lombok.Data;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @description: 延迟队列
+ * @author: wangyl
+ * @date: 2024/6/30
+ */
+@Data
+public class DelayedQueue implements Delayed {
+ /**
+ * 延迟时间
+ */
+ private long time;
+ /**
+ * 延迟任务名称
+ */
+ private String name;
+ /**
+ * 延迟任务创建时间
+ */
+ private String timestamp = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
+
+
+ public DelayedQueue(String name, long time, TimeUnit unit) {
+ this.name = name;
+ this.time = System.currentTimeMillis() + (time > 0 ? unit.toMillis(time) : 0);
+ }
+
+ @Override
+ public long getDelay(TimeUnit unit) {
+ return time - System.currentTimeMillis();
+ }
+
+ @Override
+ public int compareTo(Delayed o) {
+ DelayedQueue Order = (DelayedQueue) o;
+ long diff = this.time - Order.time;
+ if (diff <= 0) {
+ return -1;
+ } else {
+ return 1;
+ }
+ }
+
+ public static void main(String[] args) throws InterruptedException {
+ DelayedQueue Order1 = new DelayedQueue("Order1", 5, TimeUnit.SECONDS);
+ DelayedQueue Order2 = new DelayedQueue("Order2", 10, TimeUnit.SECONDS);
+ DelayedQueue Order3 = new DelayedQueue("Order3", 15, TimeUnit.SECONDS);
+ DelayQueue delayQueue = new DelayQueue<>();
+ delayQueue.put(Order1);
+ delayQueue.put(Order2);
+ delayQueue.put(Order3);
+
+ System.out.println("订单延迟队列开始时间:" + LocalDateTime.now()
+ .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
+ while (delayQueue.size() != 0) {
+ /**
+ * 取队列头部元素是否过期
+ */
+ DelayedQueue take = delayQueue.take();
+ System.out.format("订单:{%s}被取消, 放入时间{%s},取消时间:{%s}\n", take.getName(), take.getTimestamp(), LocalDateTime.now()
+ .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
+ Thread.sleep(1000);
+ }
+ }
+}
\ No newline at end of file
diff --git a/javaBase/src/main/java/concurrent/reentrantlock/AOSLockTest.java b/javaBase/src/main/java/concurrent/reentrantlock/AOSLockTest.java
new file mode 100644
index 0000000..19aaa0d
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/reentrantlock/AOSLockTest.java
@@ -0,0 +1,99 @@
+package concurrent.reentrantlock;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * ReentrantLock 公平锁运行
+ *
+ * @author hz21056617
+ * @version V1.0
+ * @ClassName: ReentrantLockNonFairTest
+ * @Date: 2022/1/28 13:48
+ */
+@Data
+@AllArgsConstructor
+public class AOSLockTest {
+ volatile static AtomicInteger num = new AtomicInteger(0);
+ static int threadNum = 3;
+
+ @AllArgsConstructor
+ static public class ReentrantLockFair implements Runnable {
+ static final ReentrantLock lock = new ReentrantLock(true);
+ Integer threadId;
+
+ @Override
+ public void run() {
+ int i = num.incrementAndGet();
+ try {
+ if (i == threadNum) {
+ TimeUnit.SECONDS.sleep(10L);
+ }
+ System.out.println("这是公平锁第" + threadId + "号线程");
+ while (true) {
+ TimeUnit.SECONDS.sleep(1L);
+ }
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ } finally {
+ lock.unlock();
+ }
+
+ }
+ }
+
+ @AllArgsConstructor
+ static public class ReentrantLockNonFair implements Runnable {
+ static final ReentrantLock lock = new ReentrantLock(false);
+ Integer threadId;
+
+ @Override
+ public void run() {
+ int i = num.incrementAndGet();
+ try {
+ if (i == threadNum) {
+ TimeUnit.SECONDS.sleep(10L);
+ }
+ lock.lock();
+ while (true) {
+ TimeUnit.SECONDS.sleep(1L);
+ }
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ } finally {
+ lock.unlock();
+ }
+ }
+ }
+
+ public static void main(String[] args) throws InterruptedException {
+// /**
+// * 公平测试
+// */
+// for (int i = 0; i < threadNum; i++) {
+// ReentrantLockFair reentrantLockFair = new ReentrantLockFair(i);
+// new Thread(reentrantLockFair).start();
+// TimeUnit.SECONDS.sleep(2L);
+// }
+ /**
+ * 非公平测试 插入线程的时候AQS 队列里没有线程 threadNum=2
+ */
+ for (int i = 0; i < threadNum; i++) {
+ ReentrantLockNonFair reentrantLockNonFair = new ReentrantLockNonFair(i);
+ new Thread(reentrantLockNonFair).start();
+ TimeUnit.SECONDS.sleep(2L);
+ }
+ /**
+ * 非公平测试 插入线程的时候AQS 队列里有1个线程 threadNum=3
+ */
+// for (int i = 0; i < threadNum; i++) {
+// ReentrantLockNonFair reentrantLockNonFair = new ReentrantLockNonFair(i);
+// new Thread(reentrantLockNonFair).start();
+// TimeUnit.SECONDS.sleep(2L);
+// }
+ }
+}
diff --git a/javaBase/src/main/java/concurrent/reentrantlock/CLHLock.java b/javaBase/src/main/java/concurrent/reentrantlock/CLHLock.java
new file mode 100644
index 0000000..05e25bb
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/reentrantlock/CLHLock.java
@@ -0,0 +1,114 @@
+package concurrent.reentrantlock;// CLHLock.java
+
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * @description: CLH 锁
+ * @author: wangyl
+ * @date: 2024/6/26
+ */
+public class CLHLock {
+ /**
+ * CLH锁节点
+ */
+ private static class CLHNode {
+ /**
+ * 锁状态:默认为false,表示线程没有获取到锁;true表示线程获取到锁或正在等待.
+ * 为了保证locked状态是线程间可见的,因此用volatile关键字修饰
+ */
+ volatile boolean locked = false;
+ }
+
+ /**
+ * 尾结点,总是指向最后一个CLHNode节点
+ * 【注意】这里用了java的原子系列之AtomicReference,能保证原子更新
+ */
+ private final AtomicReference tailNode;
+ /**
+ * 当前节点的前继节点
+ */
+ private final ThreadLocal predNode;
+ /**
+ * 当前节点
+ */
+ private final ThreadLocal curNode;
+
+ /**
+ * CLHLock构造函数,用于新建CLH锁节点时做一些初始化逻辑
+ */
+ public CLHLock() {
+ /**
+ * 初始化时尾结点指向一个空的CLH节点
+ */
+ tailNode = new AtomicReference<>(new CLHNode());
+ /**
+ * 初始化当前的CLH节点
+ */
+ curNode = new ThreadLocal() {
+ @Override
+ protected CLHNode initialValue() {
+ return new CLHNode();
+ }
+ };
+ /**
+ * 初始化前继节点,注意此时前继节点没有存储CLHNode对象,存储的是null
+ */
+ predNode = new ThreadLocal();
+ }
+
+ /**
+ * 获取锁
+ */
+ public void lock() {
+ /**
+ * 取出当前线程ThreadLocal存储的当前节点,初始化值总是一个新建的CLHNode,locked状态为false。
+ */
+ CLHNode currNode = curNode.get();
+ System.out.println(Thread.currentThread().getId() + ":curNode:" + curNode.get());
+ System.out.println(Thread.currentThread().getId() + ":predNode:" + predNode.get());
+ System.out.println(Thread.currentThread().getId() + ":tailNode:" + tailNode.get());
+ /**
+ * 此时把lock状态置为true,表示一个有效状态,
+ * 即获取到了锁或正在等待锁的状态
+ */
+ currNode.locked = true;
+ /**
+ * 当一个线程到来时,总是将尾结点取出来赋值给当前线程的前继节点;
+ * 然后再把当前线程的当前节点赋值给尾节点
+ * 【注意】在多线程并发情况下,这里通过AtomicReference类能防止并发问题
+ * 【注意】哪个线程先执行到这里就会先执行predNode.set(preNode);语句,因此构建了一条逻辑线程等待链
+ * 这条链避免了线程饥饿现象发生
+ */
+ CLHNode preNode = tailNode.getAndSet(currNode);
+ // 将刚获取的尾结点(前一线程的当前节点)付给当前线程的前继节点ThreadLocal
+ // 【思考】这句代码也可以去掉吗,如果去掉有影响吗?
+ predNode.set(preNode);
+
+ // 【1】若前继节点的locked状态为false,则表示获取到了锁,不用自旋等待;
+ // 【2】若前继节点的locked状态为true,则表示前一线程获取到了锁或者正在等待,自旋等待
+ while (preNode.locked) {
+// System.out.println("线程" + Thread.currentThread().getName() + "没能获取到锁,进行自旋等待。。。");
+ }
+ // 能执行到这里,说明当前线程获取到了锁
+ System.out.println("线程" + Thread.currentThread().getName() + "获取到了锁!!!");
+ }
+
+ /**
+ * 释放锁
+ */
+ public void unLock() {
+ // 获取当前线程的当前节点
+ CLHNode node = curNode.get();
+ // 进行解锁操作
+ // 这里将locked至为false,此时执行了lock方法正在自旋等待的后继节点将会获取到锁
+ // 【注意】而不是所有正在自旋等待的线程去并发竞争锁
+ node.locked = false;
+ System.out.println("线程" + Thread.currentThread().getName() + "释放了锁!!!");
+ // 小伙伴们可以思考下,下面两句代码的作用是什么??
+ CLHNode newCurNode = new CLHNode();
+ curNode.set(newCurNode);
+
+ // 【优化】能提高GC效率和节省内存空间,请思考:这是为什么?
+ // curNode.set(predNode.get());
+ }
+}
\ No newline at end of file
diff --git a/javaBase/src/main/java/concurrent/reentrantlock/CLHLockTest.java b/javaBase/src/main/java/concurrent/reentrantlock/CLHLockTest.java
new file mode 100644
index 0000000..30660fc
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/reentrantlock/CLHLockTest.java
@@ -0,0 +1,27 @@
+package concurrent.reentrantlock;
+
+public class CLHLockTest {
+ private static int cnt = 0;
+
+ public static void main(String[] args) throws Exception {
+ final CLHLock lock = new CLHLock();
+
+ for (int i = 0; i < 2; i++) {
+ new Thread(() -> {
+ lock.lock();
+ cnt++;
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ lock.unLock();
+ }).start();
+ }
+ // 让main线程休眠10秒,确保其他线程全部执行完
+ Thread.sleep(10000);
+ System.out.println();
+ System.out.println("cnt----------->>>" + cnt);
+
+ }
+}
\ No newline at end of file
diff --git a/javaBase/src/main/java/concurrent/reentrantlock/ReentrantLockTest.java b/javaBase/src/main/java/concurrent/reentrantlock/ReentrantLockTest.java
new file mode 100644
index 0000000..dca28f0
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/reentrantlock/ReentrantLockTest.java
@@ -0,0 +1,133 @@
+package concurrent.reentrantlock;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.SneakyThrows;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * ReentrantLock 公平锁运行
+ *
+ * @author hz21056617
+ * @version V1.0
+ * @ClassName: ReentrantLockNonFairTest
+ * @Date: 2022/1/28 13:48
+ */
+@Data
+@AllArgsConstructor
+public class ReentrantLockTest {
+ private final static ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
+ private final static List data = new ArrayList();
+
+ @AllArgsConstructor
+ static public class ReentrantLockFair implements Runnable {
+ static final ReentrantLock lock = new ReentrantLock(true);
+ Integer threadId;
+
+ @Override
+ public void run() {
+ try {
+ lock.tryLock(1, TimeUnit.MINUTES);
+ System.out.println("这是公平锁第" + threadId + "号线程");
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ } finally {
+ lock.unlock();
+ }
+
+ }
+ }
+
+ @AllArgsConstructor
+ static public class ReentrantLockNonFair implements Runnable {
+ static final ReentrantLock lock = new ReentrantLock(false);
+ Integer threadId;
+
+ @Override
+ public void run() {
+ try {
+ lock.tryLock(1, TimeUnit.MINUTES);
+ System.out.println("这是非公平锁第" + threadId + "号线程");
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ } finally {
+ lock.unlock();
+ }
+ }
+ }
+
+ @AllArgsConstructor
+ static public class ReadWriteLockRead implements Runnable {
+ @SneakyThrows
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ readWriteLock.readLock().tryLock(1, TimeUnit.MINUTES);
+ System.out.println("读取列表数据");
+ data.forEach(System.out::print);
+ System.out.println("读取列表数据结束");
+ Thread.sleep(1000);
+ } finally {
+ readWriteLock.readLock().unlock();
+ }
+ }
+ }
+ }
+
+ @AllArgsConstructor
+ static public class ReadWriteLockWrite implements Runnable {
+ static AtomicInteger num = new AtomicInteger(0);
+
+ @SneakyThrows
+ @Override
+ public void run() {
+ while (true) {
+ try {
+ readWriteLock.writeLock().tryLock(1, TimeUnit.MINUTES);
+ System.out.println("写入列表数据");
+ data.add(num.getAndAdd(1));
+ System.out.println("写入列表数据结束");
+ Thread.sleep(2000);
+ } finally {
+ readWriteLock.writeLock().unlock();
+ Thread.sleep(5000);
+ }
+ }
+ }
+ }
+
+ public static void main(String[] args) {
+ /**
+ * 公平测试
+ */
+ for (int i = 0; i < 10; i++) {
+ ReentrantLockFair reentrantLockFair = new ReentrantLockFair(i);
+ new Thread(reentrantLockFair).start();
+ }
+ /**
+ * 非公平测试
+ */
+ for (int i = 0; i < 10; i++) {
+ ReentrantLockNonFair reentrantLockNonFair = new ReentrantLockNonFair(i);
+ new Thread(reentrantLockNonFair).start();
+ }
+ data.add(1);
+ ReadWriteLockRead readWriteLockRead = new ReadWriteLockRead();
+ for (int i = 0; i < 10; i++) {
+ new Thread(readWriteLockRead).start();
+ }
+ ReadWriteLockWrite readWriteLockWrite = new ReadWriteLockWrite();
+ for (int i = 0; i < 2; i++) {
+ new Thread(readWriteLockWrite).start();
+ }
+ }
+}
diff --git a/javaBase/src/main/java/concurrent/thread/SemaphoreTest.java b/javaBase/src/main/java/concurrent/thread/SemaphoreTest.java
new file mode 100644
index 0000000..0864b1a
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/thread/SemaphoreTest.java
@@ -0,0 +1,62 @@
+package concurrent.thread;
+
+import java.util.concurrent.Semaphore;
+
+/**
+ * @description: 信号量代码
+ * @author: wangyl
+ * @date: 2024/6/20
+ */
+public class SemaphoreTest {
+
+ /**
+ * 最多4个线程访问资源
+ */
+ static Semaphore semaphore = new Semaphore(4);
+
+ static class TestThread extends Thread {
+ String name;
+
+ TestThread(String name) {
+ this.name = name;
+ }
+
+ @Override
+ public void run() {
+ try {
+ System.out.println(name + " : 获取访问能力...");
+ System.out.println(name + " : 现在可以用数量: "
+ + semaphore.availablePermits());
+ semaphore.acquire();
+ System.out.println(name + " : 获取到semaphore");
+ try {
+ for (int i = 1; i <= 5; i++) {
+ System.out.println(name + " : 正在访问资源第 " + i + ", 资源可访问数量 : " + semaphore.availablePermits());
+ Thread.sleep(1000);
+ }
+ } finally {
+ System.out.println(name + " : 释放访问能力...");
+ semaphore.release();
+ }
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public static void main(String[] args) {
+ System.out.println("总共可以同时访问的线程数量 : " + semaphore.availablePermits());
+ TestThread t1 = new TestThread("A");
+ t1.start();
+ TestThread t2 = new TestThread("B");
+ t2.start();
+ TestThread t3 = new TestThread("C");
+ t3.start();
+ TestThread t4 = new TestThread("D");
+ t4.start();
+ TestThread t5 = new TestThread("E");
+ t5.start();
+ TestThread t6 = new TestThread("F");
+ t6.start();
+ }
+}
\ No newline at end of file
diff --git a/javaBase/src/main/java/concurrent/thread/SyncronizedLevelUp.java b/javaBase/src/main/java/concurrent/thread/SyncronizedLevelUp.java
new file mode 100644
index 0000000..6b67fea
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/thread/SyncronizedLevelUp.java
@@ -0,0 +1,25 @@
+//package concurrent.thread;
+//
+//import org.openjdk.jol.info.ClassLayout;
+//
+///**
+// * @description: 锁升级 TODO
+// * @author: wangyl
+// * @date: 2024/6/21
+// */
+//public class SyncronizedLevelUp {
+// public static void main(String[] args) throws InterruptedException {
+// Object o = new Object();
+// System.out.println("还没有进入到同步块");
+// System.out.println("markword:" + ClassLayout.parseInstance(o).toPrintable());
+// //默认JVM启动会有一个预热阶段,所以默认不会开启偏向锁
+// Thread.sleep(5000);
+// Object b = new Object();
+// System.out.println("还没有进入到同步块");
+// System.out.println("markword:" + ClassLayout.parseInstance(b).toPrintable());
+// synchronized (o) {
+// System.out.println("进入到了同步块");
+// System.out.println("markword:" + ClassLayout.parseInstance(o).toPrintable());
+// }
+// }
+//}
diff --git a/javaBase/src/main/java/concurrent/thread/SyncronizedTest.java b/javaBase/src/main/java/concurrent/thread/SyncronizedTest.java
new file mode 100644
index 0000000..b97f870
--- /dev/null
+++ b/javaBase/src/main/java/concurrent/thread/SyncronizedTest.java
@@ -0,0 +1,75 @@
+package concurrent.thread;
+
+/**
+ * @description: synchronized 锁对象 和 锁方法
+ * @author: wangyl
+ * @date: 2024/6/21
+ */
+public class SyncronizedTest {
+ private static Integer count = 0;
+ static Object lock = new Object();
+
+ /**
+ * @description: 锁方法
+ * @author: wangyl
+ * @date: 2024/6/21
+ */
+ static public class SynchronizedMethodRunner implements Runnable {
+
+ @Override
+ public void run() {
+ funTest();
+ }
+
+ private synchronized void funTest() {
+ try {
+ System.out.println(Thread.currentThread().getName() + ":" + (count++));
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+
+ }
+
+ /**
+ * @description: 锁对象
+ * @author: wangyl
+ * @date: 2024/6/21
+ */
+ static public class SynchronizedObjectRunner implements Runnable {
+ Object lock;
+
+ SynchronizedObjectRunner(Object lock) {
+ this.lock = lock;
+ }
+
+ @Override
+ public void run() {
+ objTest();
+ }
+
+ private void objTest() {
+ try {
+ synchronized (lock) {
+ System.out.println(Thread.currentThread().getName() + ":" + (count++));
+ Thread.sleep(1000);
+ }
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+
+ }
+
+ public static void main(String[] args) {
+ SynchronizedMethodRunner synchronizedMethodRunner = new SynchronizedMethodRunner();
+ SynchronizedObjectRunner synchronizedObjectRunner = new SynchronizedObjectRunner(lock);
+ new Thread(synchronizedMethodRunner, "wyl01").start();
+ new Thread(synchronizedMethodRunner, "wyl02").start();
+ new Thread(synchronizedMethodRunner, "wyl03").start();
+ new Thread(synchronizedObjectRunner, "wyl04").start();
+ new Thread(synchronizedObjectRunner, "wyl05").start();
+ new Thread(synchronizedObjectRunner, "wyl06").start();
+ }
+}
diff --git a/concurrent/src/main/java/concurrent/ttl/context/ContextUtil.java b/javaBase/src/main/java/concurrent/ttl/context/ContextUtil.java
similarity index 100%
rename from concurrent/src/main/java/concurrent/ttl/context/ContextUtil.java
rename to javaBase/src/main/java/concurrent/ttl/context/ContextUtil.java
diff --git a/concurrent/src/main/java/concurrent/ttl/task/CallableTask.java b/javaBase/src/main/java/concurrent/ttl/task/CallableTask.java
similarity index 100%
rename from concurrent/src/main/java/concurrent/ttl/task/CallableTask.java
rename to javaBase/src/main/java/concurrent/ttl/task/CallableTask.java
diff --git a/concurrent/src/main/java/concurrent/ttl/task/RunnableTask.java b/javaBase/src/main/java/concurrent/ttl/task/RunnableTask.java
similarity index 100%
rename from concurrent/src/main/java/concurrent/ttl/task/RunnableTask.java
rename to javaBase/src/main/java/concurrent/ttl/task/RunnableTask.java
diff --git a/javaBase/src/test/java/concurrent/completableFuture/completableFutureTest.java b/javaBase/src/test/java/concurrent/completableFuture/completableFutureTest.java
new file mode 100644
index 0000000..d888c46
--- /dev/null
+++ b/javaBase/src/test/java/concurrent/completableFuture/completableFutureTest.java
@@ -0,0 +1,432 @@
+package concurrent.completableFuture;
+
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Slf4j
+public class completableFutureTest {
+ @Test
+ public void supplyAsync() throws ExecutionException, InterruptedException {
+ CompletableFuture cf = CompletableFuture.supplyAsync(() -> {
+ System.out.println("do something....");
+ return "result";
+ });
+
+ //等待任务执行完成
+ System.out.println("结果->" + cf.get());
+ }
+
+ @Test
+ public void supplyAsyncExecutorService() throws ExecutionException, InterruptedException {
+ // 自定义线程池
+ ExecutorService executorService = Executors.newSingleThreadExecutor();
+ CompletableFuture cf = CompletableFuture.supplyAsync(() -> {
+ System.out.println("do something....");
+ return "result";
+ }, executorService);
+
+ //等待子任务执行完成
+ System.out.println("结果->" + cf.get());
+ }
+
+ @Test
+ public void runAsync() throws ExecutionException, InterruptedException {
+ CompletableFuture cf = CompletableFuture.runAsync(() -> {
+ System.out.println("do something....");
+ });
+
+ //等待任务执行完成
+ System.out.println("结果->" + cf.get());
+ }
+
+ @Test
+ public void runAsyncExecutorService() throws ExecutionException, InterruptedException {
+ // 自定义线程池
+ ExecutorService executorService = Executors.newSingleThreadExecutor();
+ CompletableFuture cf = CompletableFuture.runAsync(() -> {
+ System.out.println("do something....");
+ }, executorService);
+
+ //等待任务执行完成
+ System.out.println("结果->" + cf.get());
+ }
+
+ @Test
+ public void thenApply() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ return 1;
+ });
+
+ CompletableFuture cf2 = cf1.thenApplyAsync((result) -> {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ result += 2;
+ return result;
+ });
+ //等待任务1执行完成
+ System.out.println("cf1结果->" + cf1.get());
+ //等待任务2执行完成
+ System.out.println("cf2结果->" + cf2.get());
+ }
+
+ @Test
+ public void thenApplyAsync() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ return 1;
+ });
+
+ CompletableFuture cf2 = cf1.thenApply((result) -> {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ result += 2;
+ return result;
+ });
+ //等待任务1执行完成
+ System.out.println("cf1结果->" + cf1.get());
+ //等待任务2执行完成
+ System.out.println("cf2结果->" + cf2.get());
+ }
+
+ @Test
+ public void thenAccept() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ return 1;
+ });
+
+ CompletableFuture cf2 = cf1.thenAccept((result) -> {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ });
+
+ //等待任务1执行完成
+ System.out.println("cf1结果->" + cf1.get());
+ //等待任务2执行完成
+ System.out.println("cf2结果->" + cf2.get());
+ }
+
+ @Test
+ public void thenAcceptAsync() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ return 1;
+ });
+
+ CompletableFuture cf2 = cf1.thenAcceptAsync((result) -> {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ });
+
+ //等待任务1执行完成
+ System.out.println("cf1结果->" + cf1.get());
+ //等待任务2执行完成
+ System.out.println("cf2结果->" + cf2.get());
+ }
+
+ @Test
+ public void thenRun() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ return 1;
+ });
+
+ CompletableFuture cf2 = cf1.thenRun(() -> {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ });
+
+ //等待任务1执行完成
+ System.out.println("cf1结果->" + cf1.get());
+ //等待任务2执行完成
+ System.out.println("cf2结果->" + cf2.get());
+ }
+
+ @Test
+ public void thenRunAsync() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ return 1;
+ });
+
+ CompletableFuture cf2 = cf1.thenRunAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ });
+
+ //等待任务1执行完成
+ System.out.println("cf1结果->" + cf1.get());
+ //等待任务2执行完成
+ System.out.println("cf2结果->" + cf2.get());
+ }
+
+ @Test
+ public void whenComplete() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ int a = 1 / 0;
+ return 1;
+ });
+
+ CompletableFuture cf2 = cf1.whenComplete((result, e) -> {
+ System.out.println("上个任务结果:" + result);
+ System.out.println("上个任务抛出异常:" + e);
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ });
+
+// //等待任务1执行完成
+ System.out.println("cf1结果->" + cf1.get());
+// //等待任务2执行完成
+ System.out.println("cf2结果->" + cf2.get());
+ }
+
+ @Test
+ public void handle() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ // int a = 1/0;
+ return 1;
+ });
+
+ CompletableFuture cf2 = cf1.handle((result, e) -> {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ System.out.println("上个任务结果:" + result);
+ System.out.println("上个任务抛出异常:" + e);
+ return result + 2;
+ });
+
+ //等待任务2执行完成
+ System.out.println("cf2结果->" + cf2.get());
+ }
+
+ @Test
+ public void thenCombine() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ return 1;
+ });
+
+ CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ return 2;
+ });
+
+ CompletableFuture cf3 = cf1.thenCombine(cf2, (a, b) -> {
+ System.out.println(Thread.currentThread() + " cf3 do something....");
+ return a + b;
+ });
+
+ System.out.println("cf3结果->" + cf3.get());
+ }
+
+ @Test
+ public void thenAcceptBoth() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ return 1;
+ });
+
+ CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ return 2;
+ });
+
+ CompletableFuture cf3 = cf1.thenAcceptBoth(cf2, (a, b) -> {
+ System.out.println(Thread.currentThread() + " cf3 do something....");
+ System.out.println(a + b);
+ });
+
+ System.out.println("cf3结果->" + cf3.get());
+ }
+
+ @Test
+ public void runAfterBoth() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ return 1;
+ });
+
+ CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ return 2;
+ });
+
+ CompletableFuture cf3 = cf1.runAfterBoth(cf2, () -> {
+ System.out.println(Thread.currentThread() + " cf3 do something....");
+ });
+
+ System.out.println("cf3结果->" + cf3.get());
+ }
+
+ @Test
+ public void applyToEither() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return "cf1 任务完成";
+ });
+
+ CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return "cf2 任务完成";
+ });
+
+ CompletableFuture cf3 = cf1.applyToEither(cf2, (result) -> {
+ System.out.println("接收到" + result);
+ System.out.println(Thread.currentThread() + " cf3 do something....");
+ return "cf3 任务完成";
+ });
+
+ System.out.println("cf3结果->" + cf3.get());
+ }
+
+ @Test
+ public void acceptEither() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return "cf1 任务完成";
+ });
+
+ CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return "cf2 任务完成";
+ });
+
+ CompletableFuture cf3 = cf1.acceptEither(cf2, (result) -> {
+ System.out.println("接收到" + result);
+ System.out.println(Thread.currentThread() + " cf3 do something....");
+ });
+
+ System.out.println("cf3结果->" + cf3.get());
+ }
+
+ @Test
+ public void runAfterEither() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ System.out.println("cf1 任务完成");
+ return "cf1 任务完成";
+ });
+
+ CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ System.out.println("cf2 任务完成");
+ return "cf2 任务完成";
+ });
+
+ CompletableFuture cf3 = cf1.runAfterEither(cf2, () -> {
+ System.out.println(Thread.currentThread() + " cf3 do something....");
+ System.out.println("cf3 任务完成");
+ });
+
+ System.out.println("cf3结果->" + cf3.get());
+ }
+
+ @Test
+ public void allOf() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ System.out.println("cf1 任务完成");
+ return "cf1 任务完成";
+ });
+
+ CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ int a = 1 / 0;
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ System.out.println("cf2 任务完成");
+ return "cf2 任务完成";
+ });
+
+ CompletableFuture cf3 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ System.out.println("cf3 任务完成");
+ return "cf3 任务完成";
+ });
+
+ CompletableFuture cfAll = CompletableFuture.allOf(cf1, cf2, cf3);
+ System.out.println("cfAll结果->" + cfAll.get());
+ }
+
+ @Test
+ public void anyOf() throws ExecutionException, InterruptedException {
+ CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf1 do something....");
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ System.out.println("cf1 任务完成");
+ return "cf1 任务完成";
+ });
+
+ CompletableFuture cf2 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ System.out.println("cf2 任务完成");
+ return "cf2 任务完成";
+ });
+
+ CompletableFuture cf3 = CompletableFuture.supplyAsync(() -> {
+ try {
+ System.out.println(Thread.currentThread() + " cf2 do something....");
+ Thread.sleep(3000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ System.out.println("cf3 任务完成");
+ return "cf3 任务完成";
+ });
+
+ CompletableFuture