wyl/JavaBasiceDemo
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

java基础 并发测试代码

Browse Source
This commit is contained in:
wyl
2024-07-01 00:28:32 +08:00
parent da0e11337c
commit 30d826bb8e
23 changed files with 1258 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
concurrent/pom.xml → javaBase/pom.xml
+1 -6
View File
@@ -9,7 +9,7 @@
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>concurrent</artifactId>
<artifactId>javaBase</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
@@ -20,10 +20,5 @@
<groupId>com.alibaba</groupId>
<artifactId>transmittable-thread-local</artifactId>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-core</artifactId>
</dependency>
</dependencies>
</project>
javaBase/src/main/java/concurrent/model/BlockingQueueTest.java
+29
View File
@@ -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<Integer> blockingQueue = new ArrayBlockingQueue<>(5000);
/**
* 使用LinkedBlockingDeque 是在插入和删除分别采用了putLock和takeLock，这样可以降低线程由于线程无法获取到lock而进入WAITING状态的可能性，从而提高了线程并发执行的效率
*/
BlockingQueue<Integer> 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);
}
}
concurrent/src/main/java/concurrent/model/QueueConsumer.java → javaBase/src/main/java/concurrent/model/QueueConsumer.java
View File
concurrent/src/main/java/concurrent/model/QueueProducer.java → javaBase/src/main/java/concurrent/model/QueueProducer.java
View File
concurrent/src/main/java/concurrent/model/WaitConsumer.java → javaBase/src/main/java/concurrent/model/WaitConsumer.java
View File
concurrent/src/main/java/concurrent/model/WaitProducer.java → javaBase/src/main/java/concurrent/model/WaitProducer.java
-1
View File
@@ -1,7 +1,6 @@
package concurrent.model;
import lombok.AllArgsConstructor;
import lombok.RequiredArgsConstructor;
import java.util.Queue;
import java.util.Random;
javaBase/src/main/java/concurrent/model/WaitTest.java
+24
View File
@@ -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);
}
}
javaBase/src/main/java/concurrent/pool/ThreadPoolTest.java
+110
View File
@@ -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;
}
}
}
concurrent/src/main/java/concurrent/queue/BlockingQueueTestWork.java → javaBase/src/main/java/concurrent/queue/BlockingQueueTest.java
+5 -18
View File
@@ -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<Integer> 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<Integer> 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) {
}
}
javaBase/src/main/java/concurrent/queue/DelayedQueue.java
+74
View File
@@ -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<DelayedQueue> 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);
}
}
}
javaBase/src/main/java/concurrent/reentrantlock/AOSLockTest.java
+99
View File
@@ -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);
// }
}
}
javaBase/src/main/java/concurrent/reentrantlock/CLHLock.java
+114
View File
@@ -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<CLHNode> tailNode;
/**
* 当前节点的前继节点
*/
private final ThreadLocal<CLHNode> predNode;
/**
* 当前节点
*/
private final ThreadLocal<CLHNode> 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());
}
}
javaBase/src/main/java/concurrent/reentrantlock/CLHLockTest.java
+27
View File
@@ -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);
}
}
javaBase/src/main/java/concurrent/reentrantlock/ReentrantLockTest.java
+133
View File
@@ -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<Integer> 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();
}
}
}
javaBase/src/main/java/concurrent/thread/SemaphoreTest.java
+62
View File
@@ -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();
}
}
javaBase/src/main/java/concurrent/thread/SyncronizedLevelUp.java
+25
View File
@@ -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());
// }
// }
//}
javaBase/src/main/java/concurrent/thread/SyncronizedTest.java
+75
View File
@@ -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();
}
}
concurrent/src/main/java/concurrent/ttl/context/ContextUtil.java → javaBase/src/main/java/concurrent/ttl/context/ContextUtil.java
View File
concurrent/src/main/java/concurrent/ttl/task/CallableTask.java → javaBase/src/main/java/concurrent/ttl/task/CallableTask.java
View File
concurrent/src/main/java/concurrent/ttl/task/RunnableTask.java → javaBase/src/main/java/concurrent/ttl/task/RunnableTask.java
View File
javaBase/src/test/java/concurrent/completableFuture/completableFutureTest.java
+432
View File
@@ -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<String> 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<String> cf = CompletableFuture.supplyAsync(() -> {
System.out.println("do something....");
return "result";
}, executorService);
//等待子任务执行完成
System.out.println("结果->" + cf.get());
}
@Test
public void runAsync() throws ExecutionException, InterruptedException {
CompletableFuture<Void> 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<Void> cf = CompletableFuture.runAsync(() -> {
System.out.println("do something....");
}, executorService);
//等待任务执行完成
System.out.println("结果->" + cf.get());
}
@Test
public void thenApply() throws ExecutionException, InterruptedException {
CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
return 1;
});
CompletableFuture<Integer> 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<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
return 1;
});
CompletableFuture<Integer> 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<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
return 1;
});
CompletableFuture<Void> 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<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
return 1;
});
CompletableFuture<Void> 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<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
return 1;
});
CompletableFuture<Void> 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<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
return 1;
});
CompletableFuture<Void> 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<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
int a = 1 / 0;
return 1;
});
CompletableFuture<Integer> 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<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
// int a = 1/0;
return 1;
});
CompletableFuture<Integer> 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<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
return 1;
});
CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf2 do something....");
return 2;
});
CompletableFuture<Integer> 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<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
return 1;
});
CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf2 do something....");
return 2;
});
CompletableFuture<Void> 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<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf1 do something....");
return 1;
});
CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> {
System.out.println(Thread.currentThread() + " cf2 do something....");
return 2;
});
CompletableFuture<Void> 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<String> cf1 = CompletableFuture.supplyAsync(() -> {
try {
System.out.println(Thread.currentThread() + " cf1 do something....");
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "cf1 任务完成";
});
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
try {
System.out.println(Thread.currentThread() + " cf2 do something....");
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "cf2 任务完成";
});
CompletableFuture<String> 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<String> cf1 = CompletableFuture.supplyAsync(() -> {
try {
System.out.println(Thread.currentThread() + " cf1 do something....");
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "cf1 任务完成";
});
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
try {
System.out.println(Thread.currentThread() + " cf2 do something....");
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "cf2 任务完成";
});
CompletableFuture<Void> 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<String> 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<String> 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<Void> 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<String> 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<String> 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<String> 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<Void> cfAll = CompletableFuture.allOf(cf1, cf2, cf3);
System.out.println("cfAll结果->" + cfAll.get());
}
@Test
public void anyOf() throws ExecutionException, InterruptedException {
CompletableFuture<String> 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<String> 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<String> 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<Object> cfAll = CompletableFuture.anyOf(cf1, cf2, cf3);
System.out.println("cfAll结果->" + cfAll.get());
}
}
javaBase/src/test/java/concurrent/model/ModelTest.java
+48
View File
@@ -0,0 +1,48 @@
package concurrent.model;
import org.junit.jupiter.api.Test;
import java.lang.ref.WeakReference;
import java.util.HashMap;
import java.util.Map;
public class ModelTest {
@Test
public void test() throws InterruptedException {
Map<WeakReference<Integer>, WeakReference<Integer>> map = new HashMap<>(8);
// 注意这里~
WeakReference<Integer> key = new WeakReference<>(1);
WeakReference<Integer> value = new WeakReference<>(2);
map.put(key, value);
System.out.println("put success");
Thread.sleep(1000);
System.gc();
System.out.println("get " + map.get(key).get());
}
@Test
public void test2() throws InterruptedException {
Map<WeakReference<Integer>, WeakReference<Integer>> map = new HashMap<>(8);
WeakReference<Integer> key = new WeakReference<>(1);
WeakReference<Integer> value = new WeakReference<>(777);
map.put(key, value);
System.out.println("put success");
Thread.sleep(1000);
System.gc();
System.out.println("get " + map.get(key).get());
}
@Test
public void test3() throws InterruptedException {
Map<WeakReference<Integer>, Integer> map = new HashMap<>(8);
WeakReference<Integer> key = new WeakReference<>(1);
Integer value = 777;
map.put(key, value);
System.out.println("put success");
Thread.sleep(1000);
System.gc();
System.out.println("get " + map.get(key));
}
}
concurrent/src/test/java/concurrent/ttl/TransmittableThreadLocalTest.java → javaBase/src/test/java/concurrent/ttl/TransmittableThreadLocalTest.java
View File