深入理解线程通信
前言
开发中不免会遇到需要所有子线程执行完毕通知主线程处理某些逻辑的场景。
或者是线程 A 在执行到某个条件通知线程 B 执行某个操作。
可以通过以下几种方式实现:
等待通知机制
等待通知模式是 Java 中比较经典的线程通信方式。
两个线程通过对同一对象调用等待 wait() 和通知 notify() 方法来进行通讯。
如两个线程交替打印奇偶数:
public class TwoThreadWaitNotify { private int start = 1; private boolean flag = false; public static void main(String[] args) { TwoThreadWaitNotify twoThread = new TwoThreadWaitNotify(); Thread t1 = new Thread(new OuNum(twoThread)); t1.setName("A"); Thread t2 = new Thread(new JiNum(twoThread)); t2.setName("B"); t1.start(); t2.start(); } /** * 偶数线程 */ public static class OuNum implements Runnable { private TwoThreadWaitNotify number; public OuNum(TwoThreadWaitNotify number) { this.number = number; } @Override public void run() { while (number.start <= 100) { synchronized (TwoThreadWaitNotify.class) { System.out.println("偶数线程抢到锁了"); if (number.flag) { System.out.println(Thread.currentThread().getName() + "+-+偶数" + number.start); number.start++; number.flag = false; TwoThreadWaitNotify.class.notify(); }else { try { TwoThreadWaitNotify.class.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } } } } } /** * 奇数线程 */ public static class JiNum implements Runnable { private TwoThreadWaitNotify number; public JiNum(TwoThreadWaitNotify number) { this.number = number; } @Override public void run() { while (number.start <= 100) { synchronized (TwoThreadWaitNotify.class) { System.out.println("奇数线程抢到锁了"); if (!number.flag) { System.out.println(Thread.currentThread().getName() + "+-+奇数" + number.start); number.start++; number.flag = true; TwoThreadWaitNotify.class.notify(); }else { try { TwoThreadWaitNotify.class.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } } } } } }
输出结果:
t2+-+奇数93 t1+-+偶数94 t2+-+奇数95 t1+-+偶数96 t2+-+奇数97 t1+-+偶数98 t2+-+奇数99 t1+-+偶数100
这里的线程 A 和线程 B 都对同一个对象 TwoThreadWaitNotify.class 获取锁,A 线程调用了同步对象的 wait() 方法释放了锁并进入 WAITING 状态。
B 线程调用了 notify() 方法,这样 A 线程收到通知之后就可以从 wait() 方法中返回。
这里利用了 TwoThreadWaitNotify.class 对象完成了通信。
有一些需要注意:
wait() 、nofify() 、nofityAll() 调用的前提都是获得了对象的锁(也可称为对象监视器)。
调用 wait() 方法后线程会释放锁,进入 WAITING 状态,该线程也会被移动到等待队列中。
调用 notify() 方法会将等待队列中的线程移动到同步队列中,线程状态也会更新为 BLOCKED
从 wait() 方法返回的前提是调用 notify() 方法的线程释放锁,wait() 方法的线程获得锁。
等待通知有着一个经典范式:
线程 A 作为消费者:
获取对象的锁。
进入 while(判断条件),并调用 wait() 方法。
当条件满足跳出循环执行具体处理逻辑。
线程 B 作为生产者:
获取对象锁。
更改与线程 A 共用的判断条件。
调用 notify() 方法。
伪代码如下:
//Thread A synchronized(Object){ while(条件){ Object.wait(); } //do something } //Thread B synchronized(Object){ 条件=false;//改变条件 Object.notify(); }
join() 方法
private static void join() throws InterruptedException { Thread t1 = new Thread(new Runnable() { @Override public void run() { LOGGER.info("running"); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } } }) ; Thread t2 = new Thread(new Runnable() { @Override public void run() { LOGGER.info("running2"); try { Thread.sleep(4000); } catch (InterruptedException e) { e.printStackTrace(); } } }) ; t1.start(); t2.start(); //等待线程1终止 t1.join(); //等待线程2终止 t2.join(); LOGGER.info("main over"); }
输出结果:
2018-03-16 20:21:30.967 [Thread-1] INFO c.c.actual.ThreadCommunication - running2 2018-03-16 20:21:30.967 [Thread-0] INFO c.c.actual.ThreadCommunication - running 2018-03-16 20:21:34.972 [main] INFO c.c.actual.ThreadCommunication - main over
在 t1.join() 时会一直阻塞到 t1 执行完毕,所以最终主线程会等待 t1 和 t2 线程执行完毕。
其实从源码可以看出,join() 也是利用的等待通知机制:
核心逻辑:
while (isAlive()) { wait(0); }
在 join 线程完成后会调用 notifyAll() 方法,是在 JVM 实现中调用,所以这里看不出来。
volatile 共享内存
因为 Java 是采用共享内存的方式进行线程通信的,所以可以采用以下方式用主线程关闭 A 线程:
public class Volatile implements Runnable{ private static volatile boolean flag = true ; @Override public void run() { while (flag){ System.out.println(Thread.currentThread().getName() + "正在运行。。。"); } System.out.println(Thread.currentThread().getName() +"执行完毕"); } public static void main(String[] args) throws InterruptedException { Volatile aVolatile = new Volatile(); new Thread(aVolatile,"thread A").start(); System.out.println("main 线程正在运行") ; TimeUnit.MILLISECONDS.sleep(100) ; aVolatile.stopThread(); } private void stopThread(){ flag = false ; } }
输出结果:
thread A正在运行。。。 thread A正在运行。。。 thread A正在运行。。。 thread A正在运行。。。 thread A执行完毕
这里的 flag 存放于主内存中,所以主线程和线程 A 都可以看到。
flag 采用 volatile 修饰主要是为了内存可见性,更多内容可以查看这里。
CountDownLatch 并发工具
CountDownLatch 可以实现 join 相同的功能,但是更加的灵活。
private static void countDownLatch() throws Exception{ int thread = 3 ; long start = System.currentTimeMillis(); final CountDownLatch countDown = new CountDownLatch(thread); for (int i= 0 ;i<thread ; i++){ new Thread(new Runnable() { @Override public void run() { LOGGER.info("thread run"); try { Thread.sleep(2000); countDown.countDown(); LOGGER.info("thread end"); } catch (InterruptedException e) { e.printStackTrace(); } } }).start(); } countDown.await(); long stop = System.currentTimeMillis(); LOGGER.info("main over total time={}",stop-start); }
输出结果:
2018-03-16 20:19:44.126 [Thread-0] INFO c.c.actual.ThreadCommunication - thread run 2018-03-16 20:19:44.126 [Thread-2] INFO c.c.actual.ThreadCommunication - thread run 2018-03-16 20:19:44.126 [Thread-1] INFO c.c.actual.ThreadCommunication - thread run 2018-03-16 20:19:46.136 [Thread-2] INFO c.c.actual.ThreadCommunication - thread end 2018-03-16 20:19:46.136 [Thread-1] INFO c.c.actual.ThreadCommunication - thread end 2018-03-16 20:19:46.136 [Thread-0] INFO c.c.actual.ThreadCommunication - thread end 2018-03-16 20:19:46.136 [main] INFO c.c.actual.ThreadCommunication - main over total time=2012
CountDownLatch 也是基于 AQS(AbstractQueuedSynchronizer) 实现的,更多实现参考 ReentrantLock 实现原理
初始化一个 CountDownLatch 时告诉并发的线程,然后在每个线程处理完毕之后调用 countDown() 方法。
该方法会将 AQS 内置的一个 state 状态 -1 。
最终在主线程调用 await() 方法,它会阻塞直到 state == 0 的时候返回。
CyclicBarrier 并发工具
private static void cyclicBarrier() throws Exception { CyclicBarrier cyclicBarrier = new CyclicBarrier(3) ; new Thread(new Runnable() { @Override public void run() { LOGGER.info("thread run"); try { cyclicBarrier.await() ; } catch (Exception e) { e.printStackTrace(); } LOGGER.info("thread end do something"); } }).start(); new Thread(new Runnable() { @Override public void run() { LOGGER.info("thread run"); try { cyclicBarrier.await() ; } catch (Exception e) { e.printStackTrace(); } LOGGER.info("thread end do something"); } }).start(); new Thread(new Runnable() { @Override public void run() { LOGGER.info("thread run"); try { Thread.sleep(5000); cyclicBarrier.await() ; } catch (Exception e) { e.printStackTrace(); } LOGGER.info("thread end do something"); } }).start(); LOGGER.info("main thread"); }
CyclicBarrier 中文名叫做屏障或者是栅栏,也可以用于线程间通信。
它可以等待 N 个线程都达到某个状态后继续运行的效果。
首先初始化线程参与者。
调用 await() 将会在所有参与者线程都调用之前等待。
直到所有参与者都调用了 await() 后,所有线程从 await() 返回继续后续逻辑。
运行结果:
2018-03-18 22:40:00.731 [Thread-0] INFO c.c.actual.ThreadCommunication - thread run 2018-03-18 22:40:00.731 [Thread-1] INFO c.c.actual.ThreadCommunication - thread run 2018-03-18 22:40:00.731 [Thread-2] INFO c.c.actual.ThreadCommunication - thread run 2018-03-18 22:40:00.731 [main] INFO c.c.actual.ThreadCommunication - main thread 2018-03-18 22:40:05.741 [Thread-0] INFO c.c.actual.ThreadCommunication - thread end do something 2018-03-18 22:40:05.741 [Thread-1] INFO c.c.actual.ThreadCommunication - thread end do something 2018-03-18 22:40:05.741 [Thread-2] INFO c.c.actual.ThreadCommunication - thread end do something
可以看出由于其中一个线程休眠了五秒,所有其余所有的线程都得等待这个线程调用 await() 。
该工具可以实现 CountDownLatch 同样的功能,但是要更加灵活。甚至可以调用 reset() 方法重置 CyclicBarrier (需要自行捕获 BrokenBarrierException 处理) 然后重新执行。
线程响应中断
public class StopThread implements Runnable { @Override public void run() { while ( !Thread.currentThread().isInterrupted()) { // 线程执行具体逻辑 System.out.println(Thread.currentThread().getName() + "运行中。。"); } System.out.println(Thread.currentThread().getName() + "退出。。"); } public static void main(String[] args) throws InterruptedException { Thread thread = new Thread(new StopThread(), "thread A"); thread.start(); System.out.println("main 线程正在运行") ; TimeUnit.MILLISECONDS.sleep(10) ; thread.interrupt(); } }
输出结果:
thread A运行中。。 thread A运行中。。 thread A退出。。
可以采用中断线程的方式来通信,调用了 thread.interrupt() 方法其实就是将 thread 中的一个标志属性置为了 true。
并不是说调用了该方法就可以中断线程,如果不对这个标志进行响应其实是没有什么作用(这里对这个标志进行了判断)。
但是如果抛出了 InterruptedException 异常,该标志就会被 JVM 重置为 false。
线程池 awaitTermination() 方法
如果是用线程池来管理线程,可以使用以下方式来让主线程等待线程池中所有任务执行完毕:
private static void executorService() throws Exception{ BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(10) ; ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(5,5,1, TimeUnit.MILLISECONDS,queue) ; poolExecutor.execute(new Runnable() { @Override public void run() { LOGGER.info("running"); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } } }); poolExecutor.execute(new Runnable() { @Override public void run() { LOGGER.info("running2"); try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } } }); poolExecutor.shutdown(); while (!poolExecutor.awaitTermination(1,TimeUnit.SECONDS)){ LOGGER.info("线程还在执行。。。"); } LOGGER.info("main over"); }
输出结果:
2018-03-16 20:18:01.273 [pool-1-thread-2] INFO c.c.actual.ThreadCommunication - running2 2018-03-16 20:18:01.273 [pool-1-thread-1] INFO c.c.actual.ThreadCommunication - running 2018-03-16 20:18:02.273 [main] INFO c.c.actual.ThreadCommunication - 线程还在执行。。。 2018-03-16 20:18:03.278 [main] INFO c.c.actual.ThreadCommunication - 线程还在执行。。。 2018-03-16 20:18:04.278 [main] INFO c.c.actual.ThreadCommunication - main over
使用这个 awaitTermination() 方法的前提需要关闭线程池,如调用了 shutdown() 方法。
调用了 shutdown() 之后线程池会停止接受新任务,并且会平滑的关闭线程池中现有的任务。
管道通信
public static void piped() throws IOException { //面向于字符 PipedInputStream 面向于字节 PipedWriter writer = new PipedWriter(); PipedReader reader = new PipedReader(); //输入输出流建立连接 writer.connect(reader); Thread t1 = new Thread(new Runnable() { @Override public void run() { LOGGER.info("running"); try { for (int i = 0; i < 10; i++) { writer.write(i+""); Thread.sleep(10); } } catch (Exception e) { } finally { try { writer.close(); } catch (IOException e) { e.printStackTrace(); } } } }); Thread t2 = new Thread(new Runnable() { @Override public void run() { LOGGER.info("running2"); int msg = 0; try { while ((msg = reader.read()) != -1) { LOGGER.info("msg={}", (char) msg); } } catch (Exception e) { } } }); t1.start(); t2.start(); }
输出结果:
2018-03-16 19:56:43.014 [Thread-0] INFO c.c.actual.ThreadCommunication - running 2018-03-16 19:56:43.014 [Thread-1] INFO c.c.actual.ThreadCommunication - running2 2018-03-16 19:56:43.130 [Thread-1] INFO c.c.actual.ThreadCommunication - msg=0 2018-03-16 19:56:43.132 [Thread-1] INFO c.c.actual.ThreadCommunication - msg=1 2018-03-16 19:56:43.132 [Thread-1] INFO c.c.actual.ThreadCommunication - msg=2 2018-03-16 19:56:43.133 [Thread-1] INFO c.c.actual.ThreadCommunication - msg=3 2018-03-16 19:56:43.133 [Thread-1] INFO c.c.actual.ThreadCommunication - msg=4 2018-03-16 19:56:43.133 [Thread-1] INFO c.c.actual.ThreadCommunication - msg=5 2018-03-16 19:56:43.133 [Thread-1] INFO c.c.actual.ThreadCommunication - msg=6 2018-03-16 19:56:43.134 [Thread-1] INFO c.c.actual.ThreadCommunication - msg=7 2018-03-16 19:56:43.134 [Thread-1] INFO c.c.actual.ThreadCommunication - msg=8 2018-03-16 19:56:43.134 [Thread-1] INFO c.c.actual.ThreadCommunication - msg=9
Java 虽说是基于内存通信的,但也可以使用管道通信。
需要注意的是,输入流和输出流需要首先建立连接。这样线程 B 就可以收到线程 A 发出的消息了。
实际开发中可以灵活根据需求选择最适合的线程通信方式。
号外
最近在总结一些 Java 相关的知识点,感兴趣的朋友可以一起维护。
地址: https://github.com/crossoverJie/Java-Interview
java线程与并发相关内容推荐阅读:http://www.roncoo.com/course/list.html?courseName=java
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
Docker容器可视化监控中心搭建
摘要: 一个宿主机上可以运行多个容器化应用,容器化应用运行于宿主机上,我们需要知道该容器的运行情况,包括 CPU使用率、内存占用、网络状况以及磁盘空间等等一系列信息,而且这些信息随时间变化,我们称其为时序数据 概述 一个宿主机上可以运行多个容器化应用,容器化应用运行于宿主机上,我们需要知道该容器的运行情况,包括 CPU使用率、内存占用、网络状况以及磁盘空间等等一系列信息,而且这些信息随时间变化,我们称其为时序数据,本文将实操 如何搭建一个可视化的监控中心 来收集这些承载着具体应用的容器的时序信息并可视化分析与展示! 动手了,动手了... 准备镜像 adviser:负责收集容器的随时间变化的数据 influxdb:负责存储时序数据 grafana:负责分析和展示时序数据 部署Influxdb服务 可以将其视为一个数据库服务,其确实用于存储数据。之所以选用该数据库,原因正如官网所说: Open Source Time Series DB Platform for Metrics & Events (Time Series Data) 下面我们将该服务部署起来 dockerrun-d...
- 下一篇
Spring Boot基础教程2-RESTfull API简单项目的快速搭建
Spring Boot基础教程1-Spring Tool Suite工具的安装 Spring Boot基础教程2-RESTful API简单项目的快速搭建 Spring Boot基础教程3-配置文件详解:Properties和YAML Spring Boot基础教程4-配置文件-多环境配置 Spring Boot基础教程5-日志配置-logback和log4j2 视频教程:http://www.roncoo.com/course/view/c99516ea604d4053908c1768d6deee3d 源码地址:https://github.com/roncoo/spring-boot-demo 一、搭建一个简单的RESTfull API接口项目 1. spring-boot-starter-web的引入,其依赖包的学习 2.spring-boot-devtools的引入,其依赖包的学习 3. 代码的实现 4. 运行项目 直接运行main方法或者使用maven命令: spring-boot:run 测试: http://localhost:8080/index 带参数:http://...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
-
Docker使用Oracle官方镜像安装(12C,18C,19C)
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS8编译安装MySQL8.0.19
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- SpringBoot2配置默认Tomcat设置,开启更多高级功能
- MySQL8.0.19开启GTID主从同步CentOS8
- CentOS7,8上快速安装Gitea,搭建Git服务器
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
推荐阅读
最新文章
- SpringBoot2更换Tomcat为Jetty,小型站点的福音
- Springboot2将连接池hikari替换为druid,体验最强大的数据库连接池
- CentOS8,CentOS7,CentOS6编译安装Redis5.0.7
- MySQL8.0.19开启GTID主从同步CentOS8
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- SpringBoot2整合Redis,开启缓存,提高访问速度
- Windows10,CentOS7,CentOS8安装Nodejs环境
- CentOS7编译安装Cmake3.16.3,解决mysql等软件编译问题
- SpringBoot2编写第一个Controller,响应你的http请求并返回结果
- CentOS7设置SWAP分区,小内存服务器的救世主