多线程的创建
创建线程方式一:继承Thread类
public class Demo1CreateThread extends Thread { public static void main(String[] args) throws InterruptedException { System.out.println("-----多线程创建开始-----"); // 1.创建一个线程 CreateThread createThread1 = new CreateThread(); CreateThread createThread2 = new CreateThread(); // 2.开始执行线程 注意 开启线程不是调用run方法,而是start方法 System.out.println("-----多线程创建启动-----"); createThread1.start(); createThread2.start(); System.out.println("-----多线程创建结束-----"); } static class CreateThread extends Thread { public void run() { String name = Thread.currentThread().getName(); for (int i = 0; i < 5; i++) { System.out.println(name + "打印内容是:" + i); } } } }
创建线程方式二:实现Runnable接口
public class Demo2CreateRunnable { public static void main(String[] args) { System.out.println("-----多线程创建开始-----"); Thread thread1 = new Thread(createRunnable); Thread thread2 = new Thread(createRunnable); // 2.开始执行线程 注意 开启线程不是调用run方法,而是start方法 System.out.println("-----多线程创建启动-----"); thread1.start(); thread2.start(); System.out.println("-----多线程创建结束-----"); } static class CreateRunnable implements Runnable { public void run() { String name = Thread.currentThread().getName(); for (int i = 0; i < 5; i++) { System.out.println(name + "的内容:" + i); } } } }
创建线程方式三:实现Callable接口
实现Runnable和实现Callable接口的方式基本相同,不过是后者执行call()方法有返回值,前者线程执行体run()方法无返回值,因此可以把这两种方式归为一种这种方式,与继承Thread类的方法之间的差别如下:
1、线程只是实现Runnable或实现Callable接口,还可以继承其他类。
2、这种方式下,多个线程可以共享一个target对象,非常适合多线程处理同一份资源的情形。
3、但是编程稍微复杂,如果需要访问当前线程,必须调用Thread.currentThread()方法。
4、继承Thread类的线程类不能再继承其他父类(Java单继承决定)。
注:一般推荐采用实现接口的方式来创建多线程
线程同步
当我们使用多个线程访问同一资源的时候,且多个线程中对资源有写的操作,就容易出现线程安全问题。 要解决上述多线程并发访问一个资源的安全问题,Java中提供了同步机制(synchronized)来解决。
1、同步代码块
Object lock = new Object(); //创建锁 synchronized(lock){ //可能会产生线程安全问题的代码 }
2、同步方法
//同步方法 public synchronized void method(){ //可能会产生线程安全问题的代码 }
非静态同步方法使用的是 this锁
静态同步方法使用的是当前方法所在类的字节码对象。
3、Lock锁
Lock lock = new ReentrantLock(); lock.lock(); //需要同步操作的代码 lock.unlock();
线程状态:共六种
New→Runnble→Blocked→waiting→timeWaiting→terminated
J.U.C线程池
线程是一个程序员一定会涉及到的概念,但是线程的创建和切换都是代价比较大的。所以,我们需要有一个好的方案能做到线程的复用,这就涉及到一个概念——线程池。合理的使用线程池能够带来3个很
明显的好处:
1. 降低资源消耗:通过重用已经创建的线程来降低线程创建和销毁的消耗
2. 提高响应速度:任务到达时不需要等待线程创建就可以立即执行。
3. 提高线程的可管理性:线程池可以统一管理、分配、调优和监控。
java的线程池支持主要通过ThreadPoolExecutor来实现,我们使用的ExecutorService的各种线程池策略都是基于ThreadPoolExecutor实现的,所以ThreadPoolExecutor十分重要。要弄明白各种线程池策
略,必须先弄明白ThreadPoolExecutor。
线程池状态
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); private static final int COUNT_BITS = Integer.SIZE - 3; private static final int CAPACITY = (1 << COUNT_BITS) - 1; // runState is stored in the high-order bits private static final int RUNNING = -1 << COUNT_BITS;//对应的高3位值是111 private static final int SHUTDOWN = 0 << COUNT_BITS;//对应的高3位值是000 private static final int STOP = 1 << COUNT_BITS;//对应的高3位值是001 private static final int TIDYING = 2 << COUNT_BITS;//对应的高3位值是010 private static final int TERMINATED = 3 << COUNT_BITS;//对应的高3位值是011 // Packing and unpacking ctl private static int runStateOf(int c) { return c & ~CAPACITY; } private static int workerCountOf(int c) { return c & CAPACITY; } private static int ctlOf(int rs, int wc) { return rs | wc; }
变量ctl定义为AtomicInteger ,记录了“线程池中的任务数量”和“线程池的状态”两个信息。共32位,其中高3位表示”线程池状态”,低29位表示”线程池中的任务数量”。
1、RUNNING:处于RUNNING状态的线程池能够接受新任务,以及对新添加的任务进行处理。
2、SHUTDOWN:处于SHUTDOWN状态的线程池不可以接受新任务,但是可以对已添加的任务进行处理。
3、STOP:处于STOP状态的线程池不接收新任务,不处理已添加的任务,并且会中断正在处理的任务。
4、TIDYING:当所有的任务已终止,ctl记录的”任务数量”为0,线程池会变为TIDYING状态。当线程池变为TIDYING状态时,会执行钩子函数terminated()。terminated()在ThreadPoolExecutor类中是空的,若用户想在线程池变为TIDYING时,进行相应的处理;可以通过重载terminated()函数来实现。
5、TERMINATED:线程池彻底终止的状态。
各个状态的转换如下:
我现在分析线程池参数最全的构造方法,了解其内部的参数意义
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory, RejectedExecutionHandler handler) { if (corePoolSize < 0 || maximumPoolSize <= 0 || maximumPoolSize < corePoolSize || keepAliveTime < 0) throw new IllegalArgumentException(); if (workQueue == null || threadFactory == null || handler == null) throw new NullPointerException(); this.acc = System.getSecurityManager() == null ? null : AccessController.getContext(); this.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }
1、corePoolSize
线程池中核心线程的数量(也称为线程池的基本大小)。当提交一个任务时,线程池会新建一个线程来执行任务,直到当前线程数等于corePoolSize。如果调用了线程池的prestartAllCoreThreads()方法,线程池会提前创建并启动所有基本线程。
2、maximumPoolSize
3、keepAliveTime
4、unit
5、workQueue
ArrayBlockingQueue:基于数组结构的有界阻塞队列,FIFO。
LinkedBlockingQueue:基于链表结构的有界阻塞队列,FIFO。
PriorityBlockingQueue:具有优先级别的阻塞队列。
SynchronousQueue:不存储元素的阻塞队列,每个插入操作都必须等待一个移出操作。
6、threadFactory
7、handler
RejectedExecutionHandler,线程池的拒绝策略。所谓拒绝策略,是指将任务添加到线程池中时,线程池拒绝该任务所采取的相应策略。当向线程池中提交任务时,如果此时线程池中的线程已经饱和了,而且阻塞队列也已经满了,则线程池会选择一种拒绝策略来处理该任务。
线程池提供了四种拒绝策略:
AbortPolicy:直接抛出异常,默认策略;
CallerRunsPolicy:用调用者所在的线程来执行任务;
DiscardOldestPolicy:丢弃阻塞队列中靠最前的任务,并执行当前任务;
DiscardPolicy:直接丢弃任务; 当然我们也可以实现自己的拒绝策略,例如记录日志等等,实现RejectedExecutionHandler接口即可。
四种线程池
FixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }
![](https://img2020.cnblogs.com/blog/2138872/202104/2138872-20210410175545292-1005237676.png)
案例:
public class Demo9FixedThreadPoolCase { public static void main(String[] args) throws InterruptedException { ExecutorService exec = Executors.newFixedThreadPool(3); for (int i = 0; i < 5; i++) { exec.execute(new Demo()); Thread.sleep(10); } exec.shutdown(); } static class Demo implements Runnable { @Override public void run() { String name = Thread.currentThread().getName(); for (int i = 0; i < 2; i++) { System.out.println(name + ":" + i); } } } }
SingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }
作为单一worker线程的线程池,它把corePool和maximumPoolSize均被设置为1,和FixedThreadPool一样使用的是无界队列LinkedBlockingQueue,所以带来的影响和FixedThreadPool一样。
SingleThreadExecutor只会使用单个工作线程,它可以保证认为是按顺序执行的,任何时候都不会有多于一个的任务处于活动状态。注意,如果单个线程在执行过程中因为某些错误中止,新的线程会替代它执行后续线程。
public class Demo9SingleThreadPoolCase { static int count = 0; public static void main(String[] args) throws InterruptedException { ExecutorService exec = Executors.newSingleThreadExecutor(); for (int i = 0; i < 10; i++) { exec.execute(new Demo()); Thread.sleep(5); } exec.shutdown(); } static class Demo implements Runnable { @Override public void run() { String name = Thread.currentThread().getName(); for (int i = 0; i < 2; i++) { count++; System.out.println(name + ":" + count); } } } }
CachedThreadPool
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }
它把corePool为0,maximumPoolSize为Integer.MAX_VALUE,这就意味着所有的任务一提交就会加入到阻塞队列中。因为线程池的基本大小设置为0,一般情况下线程池中没有程池,用的时候再创建。
但是这样就处理线程池会存在一个问题,如果主线程提交任务的速度远远大于CachedThreadPool的处理速度,则CachedThreadPool会不断地创建新线程来执行任务,这样有可能会导致系统耗尽CPU和内
存资源,所以在使用该线程池是,一定要注意控制并发的任务数,否则创建大量的线程可能导致严重的性能问题。
![](https://img2020.cnblogs.com/blog/2138872/202104/2138872-20210410212628362-1060793131.png)
案例:
public class Demo9CachedThreadPoolCase { public static void main(String[] args) throws InterruptedException { ExecutorService exec = Executors.newCachedThreadPool(); for (int i = 0; i < 10; i++) { exec.execute(new Demo()); Thread.sleep(1); } exec.shutdown(); } static class Demo implements Runnable { @Override public void run() { String name = Thread.currentThread().getName(); try { //修改睡眠时间,模拟线程执行需要花费的时间 Thread.sleep(1); System.out.println(name + "执行完了"); } catch (InterruptedException e) { e.printStackTrace(); } } } }
ScheduledThreadPool
Timer与TimerTask虽然可以实现线程的周期和延迟调度,但是Timer与TimerTask存在一些问题:
1、Timer在执行定时任务时只会创建一个线程,所以如果存在多个任务,且任务时间过长,超过了两个任务的间隔时间,会发生一些缺陷。
2、如果TimerTask抛出RuntimeException,Timer会停止所有任务的运行。
3、Timer执行周期任务时依赖系统时间,如果当前系统时间发生变化会出现一些执行上的变化
为了解决这些问题,我们一般都是推荐ScheduledThreadPoolExecutor来实现。
ScheduledThreadPoolExecutor,继承ThreadPoolExecutor且实现了ScheduledExecutorService接口,它就相当于提供了“延迟”和“周期执行”功能的ThreadPoolExecutor。
ScheduledThreadPoolExecutor,它可另行安排在给定的延迟后运行命令,或者定期执行命令。需要多个辅助线程时,或者要求 ThreadPoolExecutor 具有额外的灵活性或功能时,此类要优于Timer。
提供了四种构造方法:
public ScheduledThreadPoolExecutor(int corePoolSize) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue()); } public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), threadFactory); } public ScheduledThreadPoolExecutor(int corePoolSize, RejectedExecutionHandler handler) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), handler); } public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory, RejectedExecutionHandler handler) { super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS, new DelayedWorkQueue(), threadFactory, handler); }
在ScheduledThreadPoolExecutor的构造函数中,我们发现它都是利用ThreadLocalExecutor来构造的,唯一变动的地方就在于它所使用的阻塞队列变成了DelayedWorkQueue。
DelayedWorkQueue为ScheduledThreadPoolExecutor中的内部类,类似于延时队列和优先级队列。在执行定时任务的时候,每个任务的执行时间都不同,所以DelayedWorkQueue的工作就是按照执行时间的升序来排列,执行时间距离当前时间越近的任务在队列的前面,这样就可以保证每次出队的任务都是当前队列中执行时间最靠前的。
public class Demo9ScheduledThreadPool { public static void main(String[] args) throws InterruptedException { ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(2); System.out.println("程序开始:" + new Date()); // 第二个参数是延迟多久执行 scheduledThreadPool.schedule(new Task(), 0, TimeUnit.SECONDS); scheduledThreadPool.schedule(new Task(), 1, TimeUnit.SECONDS); scheduledThreadPool.schedule(new Task(), 5, TimeUnit.SECONDS); Thread.sleep(5000); // 关闭线程池 scheduledThreadPool.shutdown(); } static class Task implements Runnable { @Override public void run() { try { String name = Thread.currentThread().getName(); System.out.println(name + ", 开始:" + new Date()); Thread.sleep(1000); System.out.println(name + ", 结束:" + new Date()); } catch (InterruptedException e) { e.printStackTrace(); } } } }
在我们的日常开发中,我们偶尔会遇到在业务层中我们需要同时修改多张表的数据并且需要有序的执行,如果我们用往常的同步的方式,也就是单线程的方式来执行的话,可能会出现执行超时等异常造成请求结果失败,及时成功,前端也需要等待较长时间来获取响应结果,这样不但造成了用户体验差,而且会经常出现请求执行失败的问题,在这里我们一般会采用3种方式来处理,如下所示:
在采用三种方式之前,我们所有来观察一下使用同步的方式实现的结果:
1、创建maven的jar工程,引入依赖
<parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.1.0.RELEASE</version> <relativePath/> <!-- lookup parent from repository --> </parent> <dependencies> <!-- SpringBoot 容器 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> </dependencies>
2、创建启动类
@SpringBootApplication public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } }
3、编写Controller
@Controller public class TestController { @Autowired private TestAsyncService service; /** * 使用传统方式测试 */ @GetMapping("/test") public String test() { System.out.println("获取主线程名称:" + Thread.currentThread().getName()); service.serviceTest(); System.out.println("执行成功,返回结果"); return "ok"; } }
4、编写service
@Service public class TestAsyncService { public void serviceTest() { // 这里执行实际的业务逻辑,在这里我们就是用一个简单的遍历来模拟 Arrays.stream(new int[]{1,2,3,4,5,6,7,8,9,10}).forEach(t -> { try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("获取number为:" + t) ; }); } }
5、访问http://localhost:8080/test,结果如下:
使用线程池的方式来实现
修改Controller@Controller public class TestController { @Autowired private TestAsyncService service; /** * 使用传统方式测试 */ @GetMapping("/test") public String test() { System.out.println("获取主线程名称:" + Thread.currentThread().getName()); /** * 这里也可以采用以下方式使用,但是使用线程池的方式可以很便捷的对线程管理(提高程序的整体性能), * 也可以减少每次执行该请求时都需要创建一个线程造成的性能消耗 * new Thread(() ->{ * run方法中的业务逻辑 * }) */ /** * 定义一个线程池 * 核心线程数(corePoolSize):1 * 最大线程数(maximumPoolSize): 1 * 保持连接时间(keepAliveTime):50000 * 时间单位 (TimeUnit):TimeUnit.MILLISECONDS(毫秒) * 阻塞队列 new LinkedBlockingQueue<Runnable>() */ ThreadPoolExecutor executor = new ThreadPoolExecutor(1,5,50000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); // 执行业务逻辑方法serviceTest() executor.execute(new Runnable() { @Override public void run() { service.serviceTest(); } }); System.out.println("执行完成,向用户响应成功信息"); return "ok"; } }
访问http://localhost:8080/test,结果如下:
我们发现在主线程中使用线程池中的线程来实现,程序的执行结果表明,主线程直接将结果进行了返回,然后才是线程池在执行业务逻辑,减少了请求响应时长。
虽然这样实现了我们想要的结果,但是,但是我们发现如果我们在多个请求中都需要这种异步请求,每次都写这么冗余的线程池配置,这种问题当然会被我们强大的spring所观察到,所以spring为了提升开发人员的开发效率,使用@EnableAsync来开启异步的支持,使用@Async来对某个方法进行异步执行
使用注解@EnableAsync和@Async来实现
启动类:
@SpringBootApplication @EnableScheduling @ComponentScan @EnableAsync @Configuration public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } }
线程池配置
@Configuration @EnableAsync public class TaskExecutePool { private int corePoolSize = 20;//线程池维护线程的最少数量 private int maxPoolSize =20;//线程池维护线程的最大数量 private int queueCapacity = 1000; //缓存队列 private int keepAlive = 600;//允许的空闲时间 @Bean public Executor myTaskAsyncPool() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); executor.setCorePoolSize(corePoolSize); executor.setMaxPoolSize(maxPoolSize); executor.setQueueCapacity(queueCapacity); executor.setKeepAliveSeconds(keepAlive); executor.setThreadNamePrefix("MyExecutor-"); // rejection-policy:当pool已经达到max size的时候,如何处理新任务 // CALLER_RUNS:不在新线程中执行任务,而是由调用者所在的线程来执行 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); executor.initialize(); return executor; } }
@Controller public class TestController { @Autowired private TestAsyncService service; /** * 使用传统方式测试 */ @GetMapping("/test") public String test() { System.out.println("获取主线程名称:" + Thread.currentThread().getName()); service.serviceTest(); System.out.println("执行成功,返回结果"); return "ok"; } }
Service
@Service @EnableAsync public class TestAsyncService { @Async public void serviceTest() { // 这里执行实际的业务逻辑,在这里我们就是用一个简单的遍历来模拟 Arrays.stream(new int[]{1,2,3,4,5,6,7,8,9,10}).forEach(t -> { try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("获取number为:" + t) ; }); } }
访问http://localhost:8080/test,结果如下: