自我成长 - ThreadPool
type: channel
link: https://www.yanxizhu.com/index.php/tag/ThreadPool/
items: ReentrantLock、StampedLock、ReentrantReadWriteLock、Condition
items: J.U.C之AQS
items: ScheduledThreadPool线程调度
items: Executor线程池
items: JAVA多线程4种实现方式

ReentrantLock、StampedLock、ReentrantReadWriteLock、Condition
type: item
date: 2022-04-25T16:07:45+00:00
description: 锁分为Synchronized和锁Lock。ReentrantLock与锁ReentrantLock(可重入锁)与Synchronized(同步锁)的区别：可重入性：区别不大，都是通过计数器，当计数器下降为0时，锁释放。锁的实现：Synchronized依赖JVM，ReentrantLock基于JDK实现。(类似Synchronized是操作系统实现，ReentrantLock是敲代码实现),ReentrantLock可通过看源码明白实现，Synchronized依赖JVM就看不到实现原理。性能的区别：优化之前Synchronized比ReentrantLock性能差很多，优化后Synchronized引入偏向锁、轻量级锁也就是自旋锁后，2者性能差不多了。差不多时，官方推荐Synchronized。功能的区别：Synchronized的使用方便简洁，隐式获取锁释放锁，ReentrantLock需要手动获取锁，释放锁，如果忘记容易出现死锁，所以最好再finally中释放锁。锁的颗粒度和灵活度很明显ReentrantLock优于Synchronized。ReentrantLock独有的功能：ReentrantLock可指定公平锁、非公平锁，而Synchronized只能时非公平锁。公平锁：先等待的先获取锁，反之亦然。提供了一个Condition类，可以分组唤醒需要唤醒的线程，Synchronized要么随机唤醒，要么全部唤醒。提供能够中断等待锁的线程的机制，lock.lockInterruptibly()，lockInterruptibly是一种自旋锁，通过循环来调用CAS来加锁，性能比较好，就是因为不会进入内核阻塞状态。使用场景：必须实现上面3个功能时，就必须用了。代码示例一ReentrantLock实现累加，注意与synchronized的不用用法package com.yanxizhu.demo.concurrency.lock; import lombok.extern.slf4j.Slf4j; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Semaphore; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; /** * @description: ReentrantLock锁实现计数，注意使用synchroized怎么实现的 * @author: 清风 * @date: 2022/4/25 23:09 * @version: 1.0 */ @Slf4j public class ReentrantLockDemo { //定义锁 private static Lock lock = new ReentrantLock(); //定义锁用户数 private static final int totalThread = 5000; //定义并发线程数 private static final int concurrentThread = 200; //定义总和 private static int count = 0; public static void main(String[] args) throws Exception{ //定义线程池 ExecutorService executorService = Executors.newCachedThreadPool(); //定义信号量 final Semaphore semaphore = new Semaphore(concurrentThread); //定义闭锁 final CountDownLatch countDownLatch = new CountDownLatch(totalThread); for (int i = 0; i < totalThread; i++) { executorService.execute(() -> { try { semaphore.acquire(); sum(); semaphore.release(); } catch (InterruptedException e) { log.error("错误信息:【{}】", e.getMessage()); } countDownLatch.countDown(); }); } …

J.U.C之AQS
type: item
date: 2022-04-25T14:47:12+00:00
description: AQSJ.U.C之AQS，底层由线程队列实现。AbstractQueuedSynchronizer简称AQS，JUC的核心。使用Node实现FIFO队列，可以用于构建锁或者其他同步装置的基础框架。利用了一个int类型表示状态。使用方法是继承。子类通过继承并通过实现它的方法管理其状态{acquire和release}的方法操纵状态。可以同时实现排它锁和共享锁模式（独占、共享）。AQS主要同步组件CountDownLatchSemaphoreCyclicBarrierReentrantLockConditionFutureTask依次介绍1、CountDownLatch-闭锁可以完成类似阻塞的功能。一个线程或多个线程等待，直到其它线程完成。使用场景，程序执行需要某个条件执行完后，才执行。示例一package com.yanxizhu.demo.concurrency.aqs; import com.yanxizhu.demo.concurrency.annotation.ThreadSafety; import lombok.extern.slf4j.Slf4j; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; /** * @description: CountDownLatch,实现线程阻塞，等待执行完再执行程序打印 * @author: 清风 * @date: 2022/4/25 20:24 * @version: 1.0 */ @Slf4j @ThreadSafety public class CountDownLatchDemo { private static final int threadNum = 500; public static void main(String[] args) throws InterruptedException { ExecutorService executorService = Executors.newCachedThreadPool(); CountDownLatch countDownLatch = new CountDownLatch(threadNum); for (int i = 0; i <threadNum; i++) { final int num = i; executorService.execute(() -> { try{ pringNum(num); }catch (Exception e){ log.error("执行错误【{}】", e.getMessage()); }finally { countDownLatch.countDown(); } }); } countDownLatch.await(); executorService.shutdown(); log.info("finash .."); } public static void pringNum(int num) throws InterruptedException { Thread.sleep(100); log.info("num=【{}】", num); Thread.sleep(100); } }输出结果：。。。。。 20:36:34.598 [pool-1-thread-249] INFO com.yanxizhu.demo.concurrency.aqs.CountDownLatchDemo - num=【248】 20:36:34.608 [pool-1-thread-259] INFO com.yanxizhu.demo.concurrency.aqs.CountDownLatchDemo - num=【258】 20:36:34.597 [pool-1-thread-237] INFO com.yanxizhu.demo.concurrency.aqs.CountDownLatchDemo - num=【236】 20:36:34.597 [pool-1-thread-206] INFO com.yanxizhu.demo.concurrency.aqs.CountDownLatchDemo - num=【205】 20:36:34.608 [pool-1-thread-256] INFO com.yanxizhu.demo.concurrency.aqs.CountDownLatchDemo - num=【255】 20:36:34.608 [pool-1-…

ScheduledThreadPool线程调度
type: item
date: 2022-03-19T14:32:22+00:00
description: ScheduledThreadPool线程调度package com.yanxizhu; import java.util.Random; import java.util.concurrent.Callable; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; /* * 一、线程池：提供了一个线程队列，队列中保存着所有等待状态的线程。避免了创建与销毁额外开销，提高了响应的速度。 * * 二、线程池的体系结构： * java.util.concurrent.Executor : 负责线程的使用与调度的根接口 * |--**ExecutorService 子接口: 线程池的主要接口 * |--ThreadPoolExecutor 线程池的实现类 * |--ScheduledExecutorService 子接口：负责线程的调度 * |--ScheduledThreadPoolExecutor ：继承 ThreadPoolExecutor， 实现 ScheduledExecutorService * * 三、工具类 : Executors * ExecutorService newFixedThreadPool() : 创建固定大小的线程池 * ExecutorService newCachedThreadPool() : 缓存线程池，线程池的数量不固定，可以根据需求自动的更改数量。 * ExecutorService newSingleThreadExecutor() : 创建单个线程池。线程池中只有一个线程 * * ScheduledExecutorService newScheduledThreadPool() : 创建固定大小的线程，可以延迟或定时的执行任务。 */ public class TestScheduledThreadPool { public static void main(String[] args) throws Exception { ScheduledExecutorService pool = Executors.newScheduledThreadPool(5); for (int i = 0; i < 5; i++) { Future<Integer> result = pool.schedule(new Callable<Integer>(){ @Override public Integer call() throws Exception { int num = new Random().nextInt(100);//生成随机数 System.out.println(Thread.currentThread().getName() + " : " + num); return num; } }, 1, TimeUnit.SECONDS); System.out.println(result.get()); } pool.shutdown(); } }

Executor线程池
type: item
date: 2022-03-19T14:23:00+00:00
description: Executors线程池package com.yanxizhu; import java.util.ArrayList; import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; /* * 一、线程池：提供了一个线程队列，队列中保存着所有等待状态的线程。避免了创建与销毁额外开销，提高了响应的速度。 * * 二、线程池的体系结构： * java.util.concurrent.Executor : 负责线程的使用与调度的根接口 * |--**ExecutorService 子接口: 线程池的主要接口 * |--ThreadPoolExecutor 线程池的实现类 * |--ScheduledExecutorService 子接口：负责线程的调度 * |--ScheduledThreadPoolExecutor ：继承 ThreadPoolExecutor， 实现 ScheduledExecutorService * * 三、工具类 : Executors * ExecutorService newFixedThreadPool() : 创建固定大小的线程池 * ExecutorService newCachedThreadPool() : 缓存线程池，线程池的数量不固定，可以根据需求自动的更改数量。 * ExecutorService newSingleThreadExecutor() : 创建单个线程池。线程池中只有一个线程 * * ScheduledExecutorService newScheduledThreadPool() : 创建固定大小的线程，可以延迟或定时的执行任务。 */ public class TestThreadPool { public static void main(String[] args) throws Exception { //1. 创建线程池 ExecutorService pool = Executors.newFixedThreadPool(5); List<Future<Integer>> list = new ArrayList<>(); for (int i = 0; i < 10; i++) { Future<Integer> future = pool.submit(new Callable<Integer>(){ @Override public Integer call() throws Exception { int sum = 0; for (int i = 0; i <= 100; i++) { sum += i; } return sum; } }); list.add(future); } pool.shutdown(); for (Future<Integer> future : list) { System.out.println(future.get()); } /*ThreadPoolDemo tpd = new ThreadPoolDemo(); //2. 为线程池中的线程分配任务 for (int i = 0; i < 10; i++) { pool.submit(tpd); } //3. 关闭线程池 pool.shutdown();*/ } // new Thread(tpd).start(); // new Thread(tpd).start(); } class ThreadPoolDemo implements Runnable{ private int i = 0; @Override public void run() { while(i <= 100){ System.out.println(Thread.currentThread().getName() + " : " + i++); } } }

JAVA多线程4种实现方式
type: item
date: 2022-03-10T14:04:00+00:00
description: 异步与线程池1、初始化线程的4种方式1）、继承Thread2）、实现Runnable接口3）、实现Callable接口+FutureTask（可以拿到返回结果，可以处理异常）4）、线程池方式1和方式2：主进程无法获取线程的运算结果。不适合当前场景。方式3：主进程可以获取线程的运算结果，但是不利于控制服务器中的线程资源。可以导致服务器资源耗尽。方式4：通过如下两种方式初始化线程池Executors.newFiexed ThreadPool(3); //或者 new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime,TimeUnit unit, workQueue,threadFactory,handler);方式1、继续Thread/** * @description: 继承Thread * @author: 清风 * @date: 2022/3/10 21:06 * @version: 1.0 */ public class MyThread { public static void main(String[] args) { System.out.println("启动main方法开始"); OneMyThread oneMyThread = new OneMyThread(); oneMyThread.start(); System.out.println("结束main方法"); } public static class OneMyThread extends Thread{ @Override public void run() { System.out.println("当前线程："+Thread.currentThread().getId()); int i = 10/2; System.out.println("运行结果："+i); } } }运行结果:启动main方法开始 结束main方法 当前线程：22 运行结果：5方式2、实现Runable接口/** * @description: 实现Runnable接口 * @author: 清风 * @date: 2022/3/10 21:06 * @version: 1.0 */ public class MyThread { public static void main(String[] args) { System.out.println("启动main方法开始"); TwoMyThread twoMyThread = new TwoMyThread(); new Thread(twoMyThread).start(); System.out.println("结束main方法"); } /** * 实现Runnable接口 */ public static class TwoMyThread implements Runnable { @Override public void run() { System.out.println("当前线程："+Thread.currentThread().getId()); int i = 10/2; System.out.println("运行结果："+i); } } }运行结果:启动main方法开始 结束main方法 当前线程：22 运行结果：5方式3、实现Callable泛型接口/** * @description: 继承Thread * @author: 清风 * @date: 2022/3/10 21:06 * @version: 1.0 */ public class MyThread { public static void main(String[] args) throws ExecutionException, InterruptedException { System.out.println("启动main方法开始"); FutureTask<Integer> futureTask = new FutureTask<>(new ThreeMyThread()); new Thread(futureTask).start(); //等待线程执行完成，返回结果 Integer i = futureTask.get(); System.out.println("线程执行完，返回结果:"+i); System.out.println("结束main方法"); } /** * 实现Callable<T>泛型接口 */ public static class ThreeMyThread implements Callable<Integ…