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 cfAll = CompletableFuture.anyOf(cf1, cf2, cf3); + System.out.println("cfAll结果->" + cfAll.get()); + } +} diff --git a/javaBase/src/test/java/concurrent/model/ModelTest.java b/javaBase/src/test/java/concurrent/model/ModelTest.java new file mode 100644 index 0000000..785a561 --- /dev/null +++ b/javaBase/src/test/java/concurrent/model/ModelTest.java @@ -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> map = new HashMap<>(8); + // 注意这里~ + WeakReference key = new WeakReference<>(1); + WeakReference 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> map = new HashMap<>(8); + WeakReference key = new WeakReference<>(1); + WeakReference 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, Integer> map = new HashMap<>(8); + WeakReference 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)); + } + +} diff --git a/concurrent/src/test/java/concurrent/ttl/TransmittableThreadLocalTest.java b/javaBase/src/test/java/concurrent/ttl/TransmittableThreadLocalTest.java similarity index 100% rename from concurrent/src/test/java/concurrent/ttl/TransmittableThreadLocalTest.java rename to javaBase/src/test/java/concurrent/ttl/TransmittableThreadLocalTest.java