在 Java 中，多个线程之间的通信有多种方式，具体选择取决于程序的需求。先说结论：
- 简单的状态通信：使用
volatile变量、synchronized、wait/notify。 - 复杂条件通信：使用
Condition和ReentrantLock。 - 生产者-消费者模型：使用
BlockingQueue。 - 协调多个线程完成任务：使用
CountDownLatch或CyclicBarrier。
以下是几种常用的线程通信方式：
1. 使用共享变量和 synchronized 实现简单通信
可以使用共享变量（如 volatile 变量）或对象锁来实现线程间的同步通信。例如，一个线程可以修改共享变量的值，另一个线程可以根据此值的变化采取相应的操作：
class SharedResource {private volatile boolean ready = false;public synchronized void setReady() {ready = true;notifyAll(); // 通知等待的线程}public synchronized void waitForReady() throws InterruptedException {while (!ready) {wait(); // 等待 `ready` 状态变为 true}}
}
在这种情况下，一个线程可以通过调用 setReady() 改变状态并通知等待的线程，其他线程则通过 waitForReady() 方法等待通知。
2. 使用 wait() 和 notify()
wait() 和 notify() 是 Java 中线程通信的经典方式，通常用于需要在某个条件达成后才继续执行的情况。一个线程可以调用 wait() 方法等待另一个线程发出 notify() 信号。
class Communicator {private boolean flag = false;public synchronized void doWait() throws InterruptedException {while (!flag) {wait(); // 等待 `flag` 为 true}}public synchronized void doNotify() {flag = true;notify(); // 唤醒等待的线程}
}
3. 使用 Condition 和 ReentrantLock
Condition 和 ReentrantLock 提供了更灵活的线程等待和唤醒机制，比 synchronized 更适合复杂的等待条件。在使用 Condition 时，可以创建多个等待队列以实现复杂的线程通信。
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;class TaskCoordinator {private final Lock lock = new ReentrantLock();private final Condition condition = lock.newCondition();private boolean signal = false;public void waitForSignal() throws InterruptedException {lock.lock();try {while (!signal) {condition.await(); // 等待信号}} finally {lock.unlock();}}public void sendSignal() {lock.lock();try {signal = true;condition.signalAll(); // 唤醒等待的线程} finally {lock.unlock();}}
}
4. 使用 BlockingQueue 进行生产者-消费者通信
BlockingQueue 是一种线程安全的队列，适合生产者-消费者模式，生产者线程将数据放入队列，消费者线程从队列中取数据。ArrayBlockingQueue、LinkedBlockingQueue 等实现类都适合用于此类场景。
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;class ProducerConsumer {private final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();public void produce(int data) throws InterruptedException {queue.put(data); // 放入数据}public int consume() throws InterruptedException {return queue.take(); // 取出数据}
}
5. 使用 CountDownLatch 或 CyclicBarrier
CountDownLatch用于等待一组线程完成特定任务后才继续执行，适合需要等待多个任务完成后执行的场景。CyclicBarrier则用于使一组线程在一个共同的同步点等待，适合循环同步的场景。import java.util.concurrent.CountDownLatch;class LatchExample {private final CountDownLatch latch = new CountDownLatch(3); // 等待三个线程完成public void performTask() throws InterruptedException {latch.await(); // 等待// 继续执行}public void taskCompleted() {latch.countDown(); // 完成一个任务} }实现一个并发抄表任务；
-
使用
ScheduledExecutorService来执行定时任务调度，并使用Semaphore控制最大并发量，同时使用CompletableFuture和BlockingQueue实现串口的收发及任务完成后的显示更新。以下是示例代码结构：代码结构
- Scheduler：使用
ScheduledExecutorService每15分钟触发一次抄表任务。 - Concurrency Control：使用
Semaphore控制并发量，确保每次最多10个并行抄表任务。 - Task Timeout：每个任务执行时设置超时时间，超过30秒未收到数据则超时。
- Serial Port Communication：使用
BlockingQueue发送和接收消息，模拟串口的收发过程。 - Result Display：通过收到抄表回复后更新任务状态。
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;public class MeterReadingScheduler {private static final int MAX_CONCURRENT_TASKS = 10;private static final int READ_TIMEOUT_SECONDS = 30;private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);private final ExecutorService taskExecutor = Executors.newFixedThreadPool(MAX_CONCURRENT_TASKS);private final Semaphore semaphore = new Semaphore(MAX_CONCURRENT_TASKS);private final BlockingQueue<String> serialPortQueue = new LinkedBlockingQueue<>();public void start() {scheduler.scheduleAtFixedRate(this::startMeterReadingTasks, 0, 15, TimeUnit.MINUTES);}private void startMeterReadingTasks() {// 假设 meterIds 是需要抄表的表ID列表String[] meterIds = {"Meter1", "Meter2", "Meter3", /*...*/ "Meter10"};for (String meterId : meterIds) {try {semaphore.acquire(); // 限制并发量taskExecutor.submit(() -> {try {readMeter(meterId);} finally {semaphore.release(); // 任务完成释放}});} catch (InterruptedException e) {Thread.currentThread().interrupt();System.out.println("Failed to start meter reading for " + meterId);}}}private void readMeter(String meterId) {AtomicBoolean taskCompleted = new AtomicBoolean(false);ScheduledFuture<?> timeoutFuture = scheduler.schedule(() -> {if (!taskCompleted.get()) {System.out.println("Timeout for " + meterId);}}, READ_TIMEOUT_SECONDS, TimeUnit.SECONDS);// 模拟通过串口发送抄表指令try {System.out.println("Sending read command to " + meterId);serialPortQueue.put("ReadCommand:" + meterId); // 放入发送队列模拟串口发指令// 等待串口的回复String response = serialPortQueue.poll(READ_TIMEOUT_SECONDS, TimeUnit.SECONDS);if (response != null && response.equals("Response:" + meterId)) {System.out.println("Received data for " + meterId + ", updating display.");taskCompleted.set(true);} else {System.out.println("No response for " + meterId);}} catch (InterruptedException e) {Thread.currentThread().interrupt();System.out.println("Interrupted while reading " + meterId);} finally {timeoutFuture.cancel(true); // 取消超时任务}}public static void main(String[] args) {MeterReadingScheduler scheduler = new MeterReadingScheduler();scheduler.start();}
}
代码说明
ScheduledExecutorService：每15分钟调用startMeterReadingTasks方法。- 并发控制：
Semaphore保证同一时间只有10个任务并行。 - 超时处理：每个任务有一个30秒的超时处理，超时则记录失败。
- 串口收发：
BlockingQueue模拟串口的指令发送和接收。serialPortQueue.put是指令发送，serialPortQueue.poll模拟等待响应。
优化方向
- 可以改为真正的串口通信库（如
jSerialComm），替代BlockingQueue模拟串口行为。 - 通过接口更新界面显示，用
SwingWorker或事件驱动更新每个表的完成状态。 - 可以加入日志记录，每个抄表任务的开始、结束、成功或失败状态都记录到日志文件中。