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)本身是线程安全的，但它并不能阻止在列表为空时尝试移除。
讯飞绘文
讯飞绘文：免费AI写作/AI生成文章
118 查看详情
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()
以上就是Java并发消息发送系统中的会话管理与wait/notify机制深度解析的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/897957.html
微信扫一扫 
支付宝扫一扫 
