本文深入探讨了在Java多线程环境中，如何安全有效地管理共享的SMPP会话并发送大量消息。通过分析一个常见的`wait/notify`同步误用案例，我们揭示了导致`ArrayIndexOutOfBoundsException`的根本原因。文章将详细阐述`wait/notify`机制的正确用法，并引入Java并发包中的`ReentrantLock`、`Condition`以及`BlockingQueue`等高级工具，提供一种更健壮、更清晰的解决方案，以实现发送者线程与会话守护线程之间的协同工作，确保会话状态的正确同步和消息的可靠发送。
多线程环境下共享资源同步问题分析
在并发编程中，当多个线程需要访问和修改同一个共享资源时，必须采取适当的同步措施来避免数据不一致或运行时异常。本教程将以一个发送短信的场景为例，深入探讨Java中多线程同步的常见陷阱及最佳实践。
场景描述
假设我们有一个系统，需要通过SMPP协议发送大量短信。为了提高吞吐量，我们计划使用多个线程（Sender）并行发送消息。同时，由于SMPP会话可能会因网络问题中断，我们需要一个独立的“守护”线程（SessionProducer）来负责检测会话状态并在必要时重新建立连接。SMPPSession对象是所有线程共享的，它包含了发送消息和重新绑定会话的方法。
核心需求如下：
立即学习“Java免费学习笔记（深入）”；
- 多个Sender线程并发地从一个共享的消息队列中获取消息并发送。
- Sender线程只有在SMPPSession处于“已绑定”（isBind()为true）状态时才能发送消息。
- SessionProducer线程负责监控SMPPSession的状态，如果会话未绑定，则执行reBind()操作。
- 当SessionProducer成功重新绑定会话后，所有等待的Sender线程应被唤醒并继续发送消息。
- 当会话未绑定时，Sender线程应暂停执行并等待。
原始代码及问题诊断
以下是原始实现的关键代码片段：
// SMPPSession 模拟类
public class SMPPSession {
private volatile boolean bind = false; // 使用volatile确保可见性
// ... 其他方法 ...
public synchronized int sendMessage(String msg){ /* ... */ return 1; }
public synchronized void reBind(){ /* ... */ this.bind = true; /* ... */ }
public synchronized boolean isBind(){ return this.bind; }
}
// Sender 线程
public class Sender extends Thread{
private SMPPSession smppSession;
// ... 构造函数 ...
@Override
public void run(){
while (!Client.messages.isEmpty()){ // 问题点1: 外部非同步检查
synchronized (Client.messages){ // 同步块
if (smppSession.isBind()){
final String msg = Client.messages.remove(0); // 问题点2: 可能的越界访问
smppSession.sendMessage(msg);
Client.messages.notifyAll(); // 问题点3: 通知对象不合适
} else {
try {
Client.messages.wait(); // 问题点4: 等待对象不合适
} catch (InterruptedException e) { /* ... */ }
}
}
}
}
}
// SessionProducer 线程
public class SessionProducer extends Thread{
private SMPPSession smppSession;
// ... 构造函数 ...
@Override
public void run(){
while (!Client.messages.isEmpty()){ // 问题点1: 外部非同步检查
synchronized (Client.messages){ // 同步块
if (!smppSession.isBind()){
smppSession.reBind();
Client.messages.notifyAll(); // 问题点3: 通知对象不合适
} else{
try {
Client.messages.wait(); // 问题点4: 等待对象不合适
} catch (InterruptedException e) { /* ... */ }
}
}
}
}
}
// Client 主类
public class Client {
public static final List<String> messages = new CopyOnWriteArrayList<>(); // 共享消息列表
public static void main(String[] args) {
// ... 消息填充 ...
SMPPSession smppSession = new SMPPSession();
// ... 启动 SessionProducer 和 Sender 线程 ...
}
}运行上述代码，会观察到ArrayIndexOutOfBoundsException。这是因为：
- 非同步的isEmpty()检查 (问题点1): 多个Sender线程在进入synchronized (Client.messages)块之前，都可能看到Client.messages非空。例如，列表中有6条消息，4个Sender线程都判断!Client.messages.isEmpty()为真。
- 竞争条件下的remove(0) (问题点2): 当这4个Sender线程依次进入同步块后，第一个线程成功移除了消息。但当第二个、第三个、第四个线程也尝试remove(0)时，列表可能已经变空，从而抛出ArrayIndexOutOfBoundsException。
- 不合适的同步对象 (问题点3, 4): wait()和notifyAll()被调用在Client.messages对象上。然而，Sender线程等待的条件是smppSession.isBind()，SessionProducer线程等待的条件是!smppSession.isBind()。理想情况下，wait/notify应该作用于与等待条件直接相关的共享对象，即smppSession的某个状态或一个专门的锁对象。使用Client.messages作为同步对象，使得会话状态和消息列表的同步逻辑混淆，难以理解和维护。
wait(), notify(), notifyAll() 的正确使用
wait(), notify(), notifyAll() 是Java中用于线程间协作的低级机制，它们必须在synchronized块内部调用，并且作用于同一个对象。
- wait(): 使当前线程释放它持有的锁，并进入等待状态，直到被notify()或notifyAll()唤醒，或者被中断。
- notify(): 唤醒在该对象上等待的一个任意线程。
- notifyAll(): 唤醒在该对象上等待的所有线程。
关键原则：
- 必须在同步块内调用：wait(), notify(), notifyAll() 必须在synchronized (obj)块中调用，其中obj是这些方法被调用的对象。
- 等待条件应在循环中检查：线程被唤醒后，不能假设它等待的条件已经满足。可能存在“虚假唤醒”（Spurious Wakeups）或其他线程抢先修改了条件。因此，wait()调用通常放在while循环中：while (!condition) { obj.wait(); }。
- 同步对象应与等待条件相关：选择一个与被等待/被通知条件直接相关的对象作为同步锁。
优化方案：使用ReentrantLock和Condition
Java并发包（java.util.concurrent）提供了更高级、更灵活的同步工具，如ReentrantLock和Condition，它们是synchronized和wait/notify的替代品，能提供更细粒度的控制。
1. ReentrantLock和Condition管理会话状态
我们将使用ReentrantLock来保护SMPPSession的内部状态，并使用Condition来管理线程对会话绑定状态的等待和通知。
import java.util.Random;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class SMPPSession {
private volatile boolean bind = false;
private final ReentrantLock sessionLock = new ReentrantLock();
private final Condition sessionBoundCondition = sessionLock.newCondition(); // 用于等待会话绑定
private static final Random idGenerator = new Random();
public int sendMessage(String msg) {
sessionLock.lock(); // 获取会话锁
try {
while (!bind) { // 循环检查会话是否绑定
System.out.println(Thread.currentThread().getName() + " 发现会话未绑定，等待...");
sessionBoundCondition.await(); // 释放锁并等待
}
// 会话已绑定，可以发送消息
System.out.println("Sending message: " + msg);
Thread.sleep(100); // 模拟发送耗时
return Math.abs(idGenerator.nextInt());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.err.println(Thread.currentThread().getName() + " 发送消息时被中断.");
return -1;
} finally {
sessionLock.unlock(); // 释放会话锁
}
}
public void reBind() {
sessionLock.lock(); // 获取会话锁
try {
System.out.println("Rebinding...");
Thread.sleep(1500L); // 模拟重新绑定耗时
this.bind = true;
System.out.println("Session established!");
sessionBoundCondition.signalAll(); // 通知所有等待会话绑定的线程
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.err.println("Rebind操作被中断.");
} finally {
sessionLock.unlock(); // 释放会话锁
}
}
// 提供一个方法让外部可以设置会话为未绑定状态，模拟连接断开
public void setUnbound() {
sessionLock.lock();
try {
this.bind = false;
System.out.println("SMPPSession: 会话状态设置为未绑定.");
} finally {
sessionLock.unlock();
}
}
public boolean isBind() {
sessionLock.lock(); // 读取也需要加锁，确保可见性和原子性
try {
return this.bind;
} finally {
sessionLock.unlock();
}
}
}解释：
- volatile boolean bind: volatile确保bind变量在所有线程间的可见性，但它不能保证复合操作（如check-then-act）的原子性，所以仍需锁来保护。
- ReentrantLock sessionLock: 这是一个可重入的互斥锁，用于保护SMPPSession对象的关键状态和操作。
- Condition sessionBoundCondition = sessionLock.newCondition(): Condition对象是与Lock关联的，它提供了await()（类似于wait()）和signal()/signalAll()（类似于notify()/notifyAll()）方法。
- sendMessage()方法在发送前，会先获取sessionLock。如果bind为false，则调用sessionBoundCondition.await()，当前线程会释放sessionLock并进入等待状态。
- reBind()方法在重新绑定成功后，调用sessionBoundCondition.signalAll()来唤醒所有在sessionBoundCondition上等待的线程。
2. BlockingQueue处理消息队列
对于生产者-消费者模式，Java提供了BlockingQueue接口，它是线程安全的，并且自带了等待/通知机制，无需手动实现wait/notify。
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
// Sender 线程 (使用BlockingQueue)
public class Sender extends Thread {
private final SMPPSession smppSession;
private final BlockingQueue<String> messageQueue;
private final AtomicBoolean running; // 用于控制线程生命周期
public Sender(String name, SMPPSession smppSession, BlockingQueue<String> messageQueue, AtomicBoolean running) {
this.setName(name);
this.smppSession = smppSession;
this.messageQueue = messageQueue;
this.running = running;
}
@Override
public void run() {
try {
while (running.get() || !messageQueue.isEmpty()) { // 只要还在运行或队列非空就继续
String msg = messageQueue.poll(100, TimeUnit.MILLISECONDS); // 尝试获取消息，等待100ms
if (msg == null) {
// 如果短时间内没有消息，且主程序已发出停止信号，则退出
if (!running.get() && messageQueue.isEmpty()) {
break;
}
continue; // 继续循环，等待新消息或退出信号
}
int msgId = smppSession.sendMessage(msg);
if (msgId != -1) {
System.out.println(Thread.currentThread().getName() + " sent msg and received msgId: " + msgId);
} else {
// 发送失败，可以考虑将消息重新放回队列头部进行重试
System.err.println(Thread.currentThread().getName() + " failed to send message: " + msg + ", re-adding to queue.");
messageQueue.put(msg); 
