本文深入探讨了在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 messages = new CopyOnWriteArrayList(); // 共享消息列表 public static void main(String[] args) { // ... 消息填充 ... SMPPSession smppSession = new SMPPSession(); // ... 启动 SessionProducer 和 Sender 线程 ... }}
运行上述代码，会观察到ArrayIndexOutOfBoundsException。这是因为：
九歌
九歌–人工智能诗歌写作系统
322 查看详情 非同步的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 messageQueue; private final AtomicBoolean running; // 用于控制线程生命周期 public Sender(String name, SMPPSession smppSession, BlockingQueue 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);
以上就是Java多线程环境下SMPP会话与消息发送的同步机制优化的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1032568.html
微信扫一扫 
支付宝扫一扫 
