java并发消息发送系统中的会话管理与`wait`/`notify`机制深度解析。本文将探讨如何利用java的`wait`/`notify`机制在多线程环境中实现短信批量发送与会话重连。我们将分析常见的同步问题，特别是因不当的`isempty()`检查和共享资源访问导致的`arrayindexoutofboundsexception`，并提供正确同步共享资源和管理线程状态的策略，以构建健壮的并发操作。
1. 引言：并发消息发送与会话管理挑战
在企业级应用中，批量发送短信（或其他消息）是一个常见需求。为了提高吞吐量，通常会采用多线程并发发送的策略。然而，消息发送依赖于与外部服务器（如SMSC）建立的会话（SMPPSession）。这种会话可能因网络波动、服务器重启等原因中断，此时需要一个机制来重新建立会话，并在会话重连期间暂停所有发送操作，待会话恢复后再继续。
本教程将深入探讨如何使用Java的Object.wait()和Object.notifyAll()机制来协调多个消息发送线程和一个会话管理线程，以实现上述功能。我们将分析在并发场景下可能遇到的同步问题，并提供一套健壮的解决方案。
2. wait()与notify()机制详解
wait()和notify()（或notifyAll()）是Java中用于线程间协作的基础机制，它们允许线程在特定条件下暂停执行并等待，直到另一个线程通知它条件满足。
- wait(): 当一个线程调用wait()方法时，它会释放当前持有的对象锁，并进入等待状态，直到被notify()或notifyAll()唤醒，或者被中断。
- notify(): 唤醒在该对象上等待的一个任意线程。
- notifyAll(): 唤醒在该对象上等待的所有线程。
关键点：
立即学习“Java免费学习笔记（深入）”；
- 必须在synchronized块内调用：wait()、notify()和notifyAll()方法必须在持有对象监视器（即synchronized块所锁定的对象）的情况下调用，否则会抛出IllegalMonitorStateException。
- 操作同一个监视器对象：所有等待和通知操作都必须针对同一个对象进行，这个对象充当了线程间通信的“信号量”。
- 虚假唤醒与条件检查：wait()方法可能会在没有收到通知的情况下被唤醒（虚假唤醒）。因此，wait()调用通常应该放在一个while循环中，不断检查等待的条件是否真正满足。
3. 原始代码中的同步问题分析
原始代码尝试使用Client.messages列表作为监视器对象来协调线程。然而，其中存在几个关键的同步问题，导致了ArrayIndexOutOfBoundsException和线程不同步。
3.1 while (!Client.messages.isEmpty())的竞态条件
在Sender和SessionProducer线程的run()方法中，外部的while (!Client.messages.isEmpty())循环条件是在synchronized (Client.messages)块外部检查的。
// Sender线程示例
while (!Client.messages.isEmpty()){ // 问题：在同步块外检查
synchronized (Client.messages){
// ...
}
}问题分析： 假设Client.messages中只剩一条消息。多个Sender线程可能同时执行到while (!Client.messages.isEmpty())，它们都发现列表不为空，然后都尝试进入synchronized (Client.messages)块。当第一个线程进入同步块并成功移除消息后，列表变为空。此时，后续进入同步块的线程在执行Client.messages.remove(0)时，就会因为列表已空而抛出ArrayIndexOutOfBoundsException。
3.2 remove(0)的并发访问问题
即使isEmpty()检查放在同步块内，remove(0)操作也需要谨慎。如果多个线程同时尝试移除，且消息数量不足，仍然可能导致问题。在CopyOnWriteArrayList中，remove(0)本身是线程安全的，但它并不能阻止在列表为空时尝试移除。
3.3 不当的通知机制
Sender线程在成功发送消息后调用了Client.messages.notifyAll()。然而，此时SessionProducer线程可能正在等待会话断开，或者其他Sender线程可能正在等待消息或会话。这种通知通常是不必要的，并且可能导致不必要的线程唤醒，甚至掩盖真正的等待条件。
3.4 wait()的条件检查缺失
原始代码中的wait()没有放在while循环中检查条件。
// 原始Sender线程的等待逻辑
} else {
try {
Client.messages.wait(); // 问题：没有在while循环中检查条件
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}这可能导致线程在被唤醒后，其等待的条件（例如smppSession.isBind()为true或Client.messages不为空）实际上并未满足，从而导致逻辑错误或再次进入等待状态。
4. 改进方案与最佳实践
为了解决上述问题，我们需要对代码进行重构，遵循以下核心原则：
- 统一监视器对象：选择一个能代表共享状态的唯一对象作为所有wait()和notifyAll()操作的监视器。在本例中，SMPPSession对象本身是一个很好的选择，因为它代表了会话的绑定状态。
- 所有共享资源访问都需同步：包括isEmpty()、remove()等操作，都必须在持有监视器锁的同步块内执行。
- wait()必须在while循环中检查条件：防止虚假唤醒和条件不满足时继续执行。
- 明确通知时机：只有当某个线程改变了其他线程正在等待的条件时，才调用notifyAll()。
4.1 改进SMPPSession类
SMPPSession作为共享资源，其bind状态是所有线程关注的焦点。我们可以将它作为监视器对象。
public class SMPPSession {
private boolean bind = false; // 初始状态为未绑定
private static final Random idGenerator = new Random();
public synchronized int sendMessage(String msg) { // 保持sendMessage同步
try {
Thread.sleep(100L); // 模拟发送延迟
System.out.println("Sending message: " + msg);
return Math.abs(idGenerator.nextInt());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.err.println("Message sending interrupted: " + e.getMessage());
}
return -1;
}
public synchronized void reBind() { // reBind方法也同步
try {
System.out.println("Rebinding...");
Thread.sleep(2000L); // 模拟重连延迟
this.bind = true;
System.out.println("Session established!");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.err.println("Rebinding interrupted: " + e.getMessage());
}
}
public synchronized boolean isBind() { // isBind方法也同步
return this.bind;
}
public synchronized void setBind(boolean bind) { // 允许外部设置绑定状态
this.bind = bind;
}
}4.2 改进Sender线程
Sender线程需要等待两个条件：SMPPSession已绑定，且消息队列不为空。
import java.util.concurrent.CopyOnWriteArrayList;
public class Sender extends Thread {
private SMPPSession smppSession;
private CopyOnWriteArrayList messages; // 引用共享消息列表
private volatile boolean running = true; // 控制线程生命周期
public Sender(String name, SMPPSession smppSession, CopyOnWriteArrayList messages) {
this.setName(name);
this.smppSession = smppSession;
this.messages = messages;
}
public void terminate() {
this.running = false;
// 确保线程不会无限等待，如果正在wait()，需要被中断或notify
synchronized (smppSession) {
smppSession.notifyAll();
}
}
@Override
public void run() {
while (running) {
synchronized (smppSession) { // 使用smppSession作为监视器
// 等待条件：会话未绑定 或 消息队列为空
while (!smppSession.isBind() || messages.isEmpty()) {
// 如果消息已全部发送且会话已绑定，则此Sender可以退出
if (messages.isEmpty() && smppSession.isBind()) {
System.out.println(getName() + "：所有消息已发送完毕，线程退出。");
running = false; // 标记为停止
smppSession.notifyAll(); // 通知其他可能等待的线程
break; // 跳出内部while循环
}
try {
System.out.println(getName() + "：等待中... 会话绑定状态: " + smppSession.isBind() + ", 消息队列是否为空: " + messages.isEmpty());
smppSession.wait(); // 等待在smppSession对象上
} catch (InterruptedException e) {
System.out.println(getName() + "：被中断，线程退出。");
Thread.currentThread().interrupt();
running = false; // 标记为停止
break; // 跳出内部while循环
}
}
if (!running) { // 如果在等待过程中被标记为停止，则退出外部while循环
break;
}
// 条件满足：smppSession已绑定且messages不为空
final String msg = messages.remove(0); // 安全移除消息
final int msgId = smppSession.sendMessage(msg);
System.out.println(Thread.currentThread().getName() + " 发送消息并收到ID: " + msgId + "。剩余消息数：" + messages.size());
// 发送消息后，如果消息队列变空，可能需要通知其他Sender线程退出
// 或者如果Producer在等待消息队列状态，则需要通知
// 这里暂时不需要notifyAll，因为发送消息通常不改变Producer的等待条件
// 但如果messages.isEmpty()是Producer的等待条件之一，则需要
}
// 考虑在发送消息后短暂休眠，避免发送过快
try {
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
running = false;
}
}
}
} 4.3 改进SessionProducer线程
SessionProducer线程负责在会话未绑定时进行重连，并在重连成功后通知所有Sender线程。
import java.util.concurrent.CopyOnWriteArrayList;
public class SessionProducer extends Thread {
private SMPPSession smppSession;
private CopyOnWriteArrayList messages; // 引用共享消息列表，用于判断是否还有消息需要发送
private volatile boolean running = true;
public SessionProducer(String name, SMPPSession smppSession, CopyOnWriteArrayList messages) {
this.setName(name);
this.smppSession = smppSession;
this.messages = messages;
}
public void terminate() {
this.running = false;
synchronized (smppSession) {
smppSession.notifyAll();
}
}
@Override
public void run() {
while (running) {
synchronized (smppSession) { // 使用smppSession作为监视器
// 如果会话已绑定，且所有消息都已发送完毕，则Producer可以退出
if (smppSession.isBind() && messages.isEmpty()) {
System.out.println(getName() + "：所有消息已发送完毕，会话已绑定，线程退出。");
running = false;
smppSession.notifyAll(); // 通知所有线程可以退出
break;
}
// 等待条件：会话已绑定 或 消息队列为空（如果 Producer 也需要关注消息队列状态）
// 这里主要关注会话绑定状态
while (smppSession.isBind() && !messages.isEmpty()) { // 如果会话已绑定且还有消息要发，Producer等待
try {
System.out.println(getName() + "：等待中... 会话已绑定，等待会话断开或所有消息发送完毕。");
smppSession.wait(); // 等待在smppSession对象上
} catch (InterruptedException e) {
System.out.println(getName() + "：被中断，线程退出。");
Thread.currentThread().interrupt();
running = false;
break;
}
}
if (!running) {
break;
}
// 此时，会话可能未绑定，或者消息队列为空（如果上面条件包含）
if (!smppSession.isBind()) { // 如果会话未绑定，则进行重连
smppSession.reBind();
System.out.println(Thread.currentThread().getName() 
