当前位置：首页 > article >正文

多线程之线程间通讯

article 2025/7/4 21:01:59

什么是线程间通讯

线程是cpu调度的最小单位，有自己的栈空间，可以按照既定的代码逐步执行，但是如果每个线程间都是孤立地运行，就会造成资源浪费。

如果需要多个线程按照指定的规则共同完成一个任务，那么这些线程之间就需要互相协调，这个过程被称为线程的通信。

实现线程间通讯的方式

实现线程间通讯的方式可以有很多种：等待-通知、共享内存、管道流。每种方式用不同的方法来实现。

等待-通知通讯方式

等待-通知方式是java中使用普遍的线程间通讯方式，其经典的案例是“生产者-消费者”模式。
Java语言中“等待-通知”方式的线程间通信使用对象的wait()、notify()两类方法来实现。每个java对象都有wait()、notify()两类实例方法，并且wait()、notify()方法和对象的监视器是紧密相关的。

需要注意的是，wait()、notify()两类方法在数量上不止两个。wait()、notify()两类方法不属于Thread类，而是属于Java对象实例（Object实例或者Class实例）。

对象的wait()方法

对象的wait()方法的主要作用是让当前线程阻塞并等待被唤醒。wait()方法与对象监视器紧密相关，使用wait()方法时一定要放在同步块中。调用wait()方法的线程会释放锁资源，其他需要获取同一锁资源的线程可以有机会获取锁资源进而执行临界区的代码。

Object中wait()方法有三个版本：

void wait()
当前线程调用同步对象的wait()实例方法后，将导致当前线程等待，当前线程进入同步对象的监视器的WaitSet队列中，等待被其他线程唤醒。
void wait(long timeout)
这是一个限时等待版本，导致当前的线程等待，等待被其他线程唤醒，或者指定的时间timeout用完，线程不再等待。
void wait(long timeout,int nanos)
这是一个高精度的限时等待版本，其主要作用是更精确地控制等待时间。

wait()方法的原理

对象的wait()方法的核心原理大致如下：

当线程调用了locko（某个同步锁对象）的wait()后，JVM会将当前线程加入到locko监视器的WaitSet（等待集），等待被其他线程唤醒。
当前线程会释放locko对象监视器的owner权利，让其他线程可以抢夺locko对象的监视器。
让当前线程等待，其状态变成WAITING.

线程调用locko.wait()后会一直处在WAITING状态，不会再占用CPU的时间片，也不会占用同步对象locko的监视器。

对象的notify()方法

对象的notify()方法的主要作用是唤醒在等待的线程。notify()方法与对象监视器紧密相关，调动notify()方法时也需要放在同步块中。
Object中notify()方法有三个版本：

void notify()
此方法的主要作用为：locko.notify()调用后，唤醒locko监视器等待集（WaitSet）中的第一条等待线程；被唤醒的线程进入EntryList，其状态从WAITING变成BLOCKED。
void notifyAll()
locko.notifyAll()调用后，唤醒locko监视器等待集（WaitSet）中的全部等待线程；所有被唤醒的线程进入EntryList，线程状态从WAITING变成BLOCKED。

notify()方法的原理

对象的notify()或者notifyAll()方法的核心原理大致如下：

当线程调用了locko（某个同步对象）的notify()方法后，JVM会唤醒locko监视器的WaitSet中的第一条等待线程。
当线程调用了locko的notifyAll()方法后，JVM会唤醒locko监视器WaitSet中的所有等待线程。
等待线程被唤醒后，会从监视器的WaitSet移动到EntryList，线程具备了排队抢夺监视器Owner权利的资格，其状态从WAITING变成BLOCKED。
EntryList中的线程抢夺到监视器的Owner权利之后，线程的状态从BLOCKED变成Runnable，具备了重新执行的资格。

需要在synchronized同步块的内部使用wait和notify

在调用同步对象的wait()和notify()系列方法时，“当前线程”必须拥有该对象的同步锁，也就是说，wait()和notify()系列方法需要在同步块中使用，否则JVM会抛出异常。

为什么不在同步块中使用wait()和notify()系列方法，Jvm会抛出异常呢？这要从wait()和notify()系列方法的原理说起。
wait()方法的原理：首先，JVM会释放当前线程的对象锁监视器的Owner资格；其次，JVM会将当前线程移入监视器的WaitSet队列，而这些操作都和对象锁监视器有关。
所以，wait()方法必须在synchronized同步块的内部调用。在当前线程执行wait()方法前，必须通过synchronized()方法成为对象锁的监视器的Owner。

notify()方法的原理：JVM从对象锁的监视器的WaitSet队列移动一个线程到其EntryList队列，这些操作都与对象锁的监视器有关。
所以，notify()方法必须在synchronized同步块的内部调用。在当前线程执行notify()方法前，必须通过synchronized()方法成为对象锁的监视器的Owner。

Semaphore信号量方式

信号量可以控制在同一时刻访问共享资源的线程数量，通过协调各个线程以保证共享资源的合理使用。
Semaphore维护了一组虚拟许可，它的数量可以通过构造器的参数指定。线程在访问共享资源前必须调用Semaphore的acquire()方法获得许可，如果许可数量为0，该线程就一直阻塞。线程访问完资源后，必须调用Semaphore的release()方法释放许可。

Semaphore的实现借助了同步框架AQS，Semaphore使用一个内部类Sync来实现，而Sync继承了AQS来实现，Sync有两个子类，分别对应着公平模式和非公平模式的Semaphore。

Semaphore信号量作用

Semaphore信号量只是用来限制访问一个或者多个共享资源的最大线程数，并不能解决线程安全问题。

应用场景

信号量的核心功能就是用来对资源做一定的限制，防止出现崩塌现象。最适用的应用场景那就是限流，通过限流来保护对应的资源。例如，在Spring Cloud中我们会用Hystrix来保护服务，进行熔断降级。在Hystrix中有两种模式，分别是线程池和信号量。

Semaphore中主要方法

Semaphore(permits)
构造函数，permits表示许可证数量
Semaphore(permits,fair)
构造函数，permits表示许可证数量，fair表示是否使用公平策略来维护想要获取许可的线程。
availablePermits()
acquire()
获取许可，且只获取一个许可。
acquire(permits)
获取许可，获取permits个许可。
acquireUninterruptibly()
acquireUninterruptibly(permits)
tryAcquire()
tryAcquire(permits)
tryAcquire(timeout,TimeUnit)
tryAcquire(permits,timeout,TimeUnit)
release()
释放许可，且只释放一个许可。
release(permits)
释放许可，且释放permits个许可。
drainPermits()
hasQueuedThreads()
getQueueLength()
获取队列中等待获取许可的线程数

Semaphore代码实现

public class ThreadE implements Runnable {private Semaphore sr = null;private String name = null;public ThreadE(Semaphore sr, String name) {this.name = name;this.sr = sr;}@Overridepublic void run() {try {sr.acquire();System.out.println(name + "请来购票！");Thread.sleep(200);} catch (Exception e) {} finally {System.out.println(name + "购票结束!!!!!!!!");sr.release();}}
}

测试类：

public class SemaphoreTest {public static void main(String[] args) {Semaphore sr = new Semaphore(3, false);for (int a = 0; a < 10; a++) {ThreadE threadE = new ThreadE(sr, "乘客" + a);Thread t = new Thread(threadE);t.start();}}
}

CountDownLatch倒数闩方式

CountDownLatch是一个常用的共享锁，其功能相当于一个多线程环境下的倒数门闩。CountDownLatch可以指定一个计数值，在并发环境下有线程进行减一操作，当计数值变为0之后，被await方法阻塞的线程将会唤醒。通过CountDownLatch可以实现线程间的计数同步。
CountDownLatch的使用步骤：

创建倒数闩，初始化CountDownLatch时设置倒数的总次数，比如为100
等待线程调用倒数闩的await()方法阻塞自己，等待倒数闩的计数器数值为0（倒数线程全部执行结束）
倒数线程执行完，调用CountDownLatch.countDown()方法将计数器数值减一

代码实现

public class DownImgThreadTwo implements Callable<Integer> {private CountDownLatch countDownLatch;private String url;public DownImgThreadTwo(String url, CountDownLatch countDownLatch) {this.url = url;this.countDownLatch = countDownLatch;}@Overridepublic Integer call() throws Exception {try {if (url.contains("3")) {Thread.sleep(200);} else if (url.contains("2")) {Thread.sleep(500);}System.out.println(url + "下载完成！");return 1;} catch (Exception e) {return 0;} finally {countDownLatch.countDown();}}
}

测试类：

public class DownImgTest {public static void main(String[] args) throws InterruptedException, ExecutionException {CountDownLatch countDownLatch = new CountDownLatch(3);long time = System.currentTimeMillis();for (int a = 1; a < 4; a++) {DownImgThreadTwo downImgThread = new DownImgThreadTwo("图片" + a, countDownLatch);FutureTask<Integer> futureTask = new FutureTask<>(downImgThread);Thread t = new Thread(futureTask);t.start();//特别注意，如果不使用get获取结果，则最终耗时基本等于执行时间最长的那个线程的时间；// 但是如果使用get获取结果则最终耗时基本等于每个线程耗时之和Integer integer = futureTask.get();System.out.println("下载结果：" + a);}countDownLatch.await();System.out.println("所有图片下载完成");System.out.println("下载耗时：" + String.valueOf(System.currentTimeMillis() - time));}
}

注意，使用countDownLatch时，阻塞的是调用await()方法的线程。

CyclicBarrier方式（循环栅栏）

代码实现

public class ThreadF implements Runnable {private CyclicBarrier cyclicBarrier;private String name;public ThreadF(String name, CyclicBarrier cyclicBarrier) {this.name = name;this.cyclicBarrier = cyclicBarrier;}@Overridepublic void run() {try {System.out.println(Thread.currentThread().getName() + name + "报到！");cyclicBarrier.await();System.out.println(Thread.currentThread().getName() + name + "可以走了！");} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}}
}

测试类：

public class CyclicBarrierTest {public static void main(String[] args) {CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {@Overridepublic void run() {System.out.println(Thread.currentThread().getName() + "集合！！！！！!!!!!!");}});for (int a = 1; a < 11; a++) {ThreadF threadF = new ThreadF("同学" + a, cyclicBarrier);Thread t = new Thread(threadF);t.start();}}
}

CyclicBarrier与CountDownLatch的区别：