深入探讨线程通信的机制与实现方式
在多线程编程中，线程之间的通信是一个非常重要的概念。线程通信机制不仅可以协调线程的执行顺序，还能确保线程之间的数据一致性。本文将深入探讨Java中线程通信的几种常见实现方式，包括wait/notify机制、条件变量以及更高级的工具类。
wait/notify机制
Java中的wait、notify和notifyAll是线程通信的基础方法，它们都是Object类中的方法，因此每个对象都可以作为线程通信的锁。
基本概念：
wait()：让当前线程进入等待状态，直到其他线程调用notify()或notifyAll()来唤醒它。调用wait()时，必须持有该对象的监视器锁，否则会抛出IllegalMonitorStateException。notify()：唤醒一个在此对象监视器上等待的线程。唤醒的具体线程由JVM决定。notifyAll()：唤醒所有在此对象监视器上等待的线程。
生产者-消费者模型的实现
生产者-消费者模型是多线程通信的经典案例。以下是使用wait和notify方法实现的一个简单示例：
package cn.juwatech.threadcommunication;
import java.util.LinkedList;
import java.util.Queue;
class ProducerConsumer {
private final Queue<Integer> queue = new LinkedList<>();
private final int MAX_SIZE = 5;
public void produce() throws InterruptedException {
int value = 0;
while (true) {
synchronized (this) {
// 如果队列满了，等待消费者消费
while (queue.size() == MAX_SIZE) {
wait();
}
queue.offer(value);
System.out.println("Produced " + value);
value++;
// 通知消费者
notify();
// 模拟生产时间
Thread.sleep(1000);
}
}
}
public void consume() throws InterruptedException {
while (true) {
synchronized (this) {
// 如果队列为空，等待生产者生产
while (queue.isEmpty()) {
wait();
}
int value = queue.poll();
System.out.println("Consumed " + value);
// 通知生产者
notify();
// 模拟消费时间
Thread.sleep(1000);
}
}
}
}
public class ProducerConsumerExample {
public static void main(String[] args) {
ProducerConsumer pc = new ProducerConsumer();
Thread producerThread = new Thread(() -> {
try {
pc.produce();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
Thread consumerThread = new Thread(() -> {
try {
pc.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
producerThread.start();
consumerThread.start();
}
}
在这个示例中，生产者线程负责向队列中添加数据，而消费者线程则从队列中取出数据。wait()和notify()方法用于协调生产者和消费者的运行，确保线程安全地访问共享资源。
条件变量（Condition）
Condition类提供了与wait/notify机制相似的功能，但功能更为强大。它允许线程在特定的条件下等待，并提供了更细粒度的控制。
使用ReentrantLock和Condition的生产者-消费者模型
package cn.juwatech.threadcommunication;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
class ProducerConsumerWithCondition {
private final Queue<Integer> queue = new LinkedList<>();
private final int MAX_SIZE = 5;
private final Lock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();
public void produce() throws InterruptedException {
int value = 0;
while (true) {
lock.lock();
try {
while (queue.size() == MAX_SIZE) {
notFull.await();
}
queue.offer(value);
System.out.println("Produced " + value);
value++;
notEmpty.signal();
} finally {
lock.unlock();
}
Thread.sleep(1000);
}
}
public void consume() throws InterruptedException {
while (true) {
lock.lock();
try {
while (queue.isEmpty()) {
notEmpty.await();
}
int value = queue.poll();
System.out.println("Consumed " + value);
notFull.signal();
} finally {
lock.unlock();
}
Thread.sleep(1000);
}
}
}
public class ProducerConsumerConditionExample {
public static void main(String[] args) {
ProducerConsumerWithCondition pc = new ProducerConsumerWithCondition();
Thread producerThread = new Thread(() -> {
try {
pc.produce();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
Thread consumerThread = new Thread(() -> {
try {
pc.consume();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
producerThread.start();
consumerThread.start();
}
}
在这个示例中，ReentrantLock和Condition为我们提供了更加灵活的线程通信方式。我们可以为不同的条件创建多个Condition对象，从而更精细地控制线程的执行。
高级线程通信工具类
Java并发包（java.util.concurrent）提供了一些更高级的线程通信工具类，例如CountDownLatch、CyclicBarrier、Semaphore和Exchanger。
使用CountDownLatch进行线程同步
CountDownLatch可以使一个线程等待其他线程完成一组操作后再继续执行。
package cn.juwatech.threadcommunication;
import java.util.concurrent.CountDownLatch;
public class CountDownLatchExample {
public static void main(String[] args) throws InterruptedException {
CountDownLatch latch = new CountDownLatch(3);
Thread worker1 = new Thread(new Worker(latch), "Worker-1");
Thread worker2 = new Thread(new Worker(latch), "Worker-2");
Thread worker3 = new Thread(new Worker(latch), "Worker-3");
worker1.start();
worker2.start();
worker3.start();
// 主线程等待所有工人完成任务
latch.await();
System.out.println("All workers have finished their tasks. Proceeding with main thread.");
}
}
class Worker implements Runnable {
private final CountDownLatch latch;
public Worker(CountDownLatch latch) {
this.latch = latch;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " is working...");
try {
Thread.sleep(2000); // 模拟工作时间
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
latch.countDown(); // 完成任务，减少计数
}
}
}
在这个例子中，主线程会等待所有工人线程完成任务后再继续执行。CountDownLatch提供了一个简单的机制来协调多个线程之间的执行顺序。
使用CyclicBarrier进行并行任务的同步
CyclicBarrier可以使一组线程相互等待，直到所有线程都达到一个屏障点，然后这些线程才继续执行。
package cn.juwatech.threadcommunication;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierExample {
public static void main(String[] args) {
int parties = 3;
CyclicBarrier barrier = new CyclicBarrier(parties, () -> System.out.println("All parties have reached the barrier. Let's proceed."));
for (int i = 0; i < parties; i++) {
new Thread(new Task(barrier), "Thread-" + i).start();
}
}
}
class Task implements Runnable {
private final CyclicBarrier barrier;
public Task(CyclicBarrier barrier) {
this.barrier = barrier;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " is waiting at the barrier.");
try {
barrier.await(); // 等待所有线程到达屏障
System.out.println(Thread.currentThread().getName() + " is proceeding.");
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}
}
在这个示例中，所有线程在到达屏障时都会等待，直到所有线程都到达屏障，然后继续执行。这种机制非常适合并行任务的同步。
总结
线程通信在多线程编程中扮演着重要的角色，确保了线程之间的协调与数据一致性。通过理解并掌握Java中不同的线程通信机制，从基础的wait/notify到高级的并发工具类，开发者可以构建出高效、安全的并发应用程序。
