Java 线程间的通信方式-开发者社区

一、概述

在Java中，线程之间的通信主要涉及线程之间如何交换信息或协调行动。常见的线程通信方式有以下几种：

共享内存（通过共享对象进行通信）
wait/notify机制
Lock和Condition
使用阻塞队列（BlockingQueue）
使用管道（PipedInputStream/PipedOutputStream 或 PipedReader/PipedWriter）
使用信号量（Semaphore）等同步工具
使用CountDownLatch、CyclicBarrier等同步辅助类
使用Exchanger
使用Future和Callable
使用CompletableFuture

二、详细实现与示例

2.1 共享内存（共享变量）

线程之间通过读写共享的变量来进行通信。这是最常用的一种方式，但需要同步机制来保证数据的一致性（如synchronized、volatile、Lock等）。

// 示例：使用volatile共享变量publicclassSharedVariable{privatevolatilebooleanflag=false;publicvoidsetFlag(){flag=true;}publicvoidwaitForFlag(){while(!flag){// 等待}System.out.println("Flag is true!");}}

2.2 wait()/notify()/notifyAll()机制

这是Object类提供的方法，用于线程间的等待和唤醒。必须搭配synchronized使用，因为wait/notify需要先获得对象的锁。

publicclassWaitNotifyExample{privatefinalObjectlock=newObject();privatebooleancondition=false;// 等待方publicvoidawait()throwsInterruptedException{synchronized(lock){while(!condition){lock.wait();// 释放锁并等待}// 执行后续操作System.out.println("条件满足，继续执行");}}// 通知方publicvoidsignal(){synchronized(lock){condition=true;lock.notify();// 唤醒一个等待线程// lock.notifyAll(); // 唤醒所有等待线程}}}

2.3 Lock和Condition（显式锁）

Lock提供了比synchronized更灵活的锁机制，而Condition则提供了线程间协调等待和唤醒的功能，类似于wait/notify，但可以更精细地控制。

importjava.util.concurrent.locks.*;publicclassLockConditionExample{privatefinalLocklock=newReentrantLock();privatefinalConditioncondition=lock.newCondition();privatebooleanready=false;publicvoidawait()throwsInterruptedException{lock.lock();try{while(!ready){condition.await();// 等待}System.out.println("准备就绪，开始工作");}finally{lock.unlock();}}publicvoidsignal(){lock.lock();try{ready=true;condition.signal();// 发送信号}finally{lock.unlock();}}}

2.4 阻塞队列（BlockingQueue）

BlockingQueue是一个接口，它定义了线程安全的队列，支持阻塞的插入和移除操作。当队列满时，插入操作会被阻塞；当队列空时，移除操作会被阻塞。

importjava.util.concurrent.*;publicclassBlockingQueueExample{privateBlockingQueue<String>queue=newLinkedBlockingQueue<>(10);// 生产者classProducerimplementsRunnable{@Overridepublicvoidrun(){try{Stringitem="item-"+System.currentTimeMillis();queue.put(item);// 队列满时会阻塞System.out.println("生产: "+item);}catch(InterruptedExceptione){Thread.currentThread().interrupt();}}}// 消费者classConsumerimplementsRunnable{@Overridepublicvoidrun(){try{Stringitem=queue.take();// 队列空时会阻塞System.out.println("消费: "+item);}catch(InterruptedExceptione){Thread.currentThread().interrupt();}}}}

2.5 信号量（Semaphore）

信号量用来控制同时访问某个资源的线程数量，也可以用于线程间通信，但更偏向于同步。通过acquire()和release()操作。

importjava.util.concurrent.Semaphore;publicclassSemaphoreExample{privateSemaphoresemaphore=newSemaphore(0);// 初始许可为0publicvoidwaitSignal()throwsInterruptedException{System.out.println("等待信号...");semaphore.acquire();// 获取许可，没有许可时阻塞System.out.println("收到信号，继续执行");}publicvoidsendSignal(){System.out.println("发送信号");semaphore.release();// 释放许可}}

2.6 CountDownLatch

CountDownLatch允许一个或多个线程等待其他线程完成操作；

importjava.util.concurrent.CountDownLatch;publicclassCountDownLatchExample{publicstaticvoidmain(String[]args)throwsInterruptedException{CountDownLatchlatch=newCountDownLatch(3);for(inti=0;i<3;i++){newThread(()->{System.out.println(Thread.currentThread().getName()+" 完成任务");latch.countDown();// 计数器减1}).start();}latch.await();// 等待计数器归零System.out.println("所有任务完成，继续主线程");}}

2.7 CyclicBarrier

CyclicBarrier让一组线程相互等待，到达一个公共屏障点。

importjava.util.concurrent.CyclicBarrier;publicclassCyclicBarrierExample{publicstaticvoidmain(String[]args){CyclicBarrierbarrier=newCyclicBarrier(3,()->{System.out.println("所有线程到达屏障，执行屏障动作");});for(inti=0;i<3;i++){newThread(()->{try{System.out.println(Thread.currentThread().getName()+" 到达屏障");barrier.await();// 等待其他线程System.out.println(Thread.currentThread().getName()+" 继续执行");}catch(Exceptione){e.printStackTrace();}}).start();}}}

2.8 Exchanger

Exchanger用于两个线程之间交换数据。每个线程在exchange()方法处等待，直到两个线程都调用exchange()方法，然后交换数据。

importjava.util.concurrent.Exchanger;publicclassExchangerExample{publicstaticvoidmain(String[]args){Exchanger<String>exchanger=newExchanger<>();newThread(()->{try{Stringdata="Thread1的数据";System.out.println("Thread1发送: "+data);Stringreceived=exchanger.exchange(data);System.out.println("Thread1收到: "+received);}catch(InterruptedExceptione){e.printStackTrace();}}).start();newThread(()->{try{Stringdata="Thread2的数据";System.out.println("Thread2发送: "+data);Stringreceived=exchanger.exchange(data);System.out.println("Thread2收到: "+received);}catch(InterruptedExceptione){e.printStackTrace();}}).start();}}

2.9 管道（Piped Streams）

管道是一种基于流的通信方式，一个线程向管道输出流写数据，另一个线程从管道输入流读数据。这种方式适用于线程间传输数据，但不如阻塞队列常用。

importjava.io.*;publicclassPipedStreamExample{publicstaticvoidmain(String[]args)throwsIOException{PipedOutputStreampos=newPipedOutputStream();PipedInputStreampis=newPipedInputStream(pos);// 写入线程newThread(()->{try{pos.write("Hello from pipe!".getBytes());pos.close();}catch(IOExceptione){e.printStackTrace();}}).start();// 读取线程newThread(()->{try{intdata;while((data=pis.read())!=-1){System.out.print((char)data);}pis.close();}catch(IOExceptione){e.printStackTrace();}}).start();}}

2.10 Future/Callable

通过ExecutorService提交一个Callable任务，返回一个Future对象。主线程可以通过Future.get()方法获取子线程的执行结果，这实际上也是一种线程间通信（获取结果）。

importjava.util.concurrent.*;publicclassFutureExample{publicstaticvoidmain(String[]args)throwsException{ExecutorServiceexecutor=Executors.newSingleThreadExecutor();Callable<String>task=()->{Thread.sleep(1000);return"任务结果";};Future<String>future=executor.submit(task);System.out.println("等待结果...");Stringresult=future.get();// 阻塞等待结果System.out.println("结果: "+result);executor.shutdown();}}

2.11 CompletableFuture（Java 8+）

更强大的异步编程和线程通信。

importjava.util.concurrent.CompletableFuture;publicclassCompletableFutureExample{publicstaticvoidmain(String[]args){// 链式调用和组合CompletableFuture<String>future=CompletableFuture.supplyAsync(()->"Hello").thenApply(s->s+" World").thenApply(String::toUpperCase);future.thenAccept(System.out::println);// 结果: HELLO WORLD}}