Java并发消息发送系统中的会话管理与wait/notify机制深度解析

// Sender线程示例
while (!Client.messages.isEmpty()){ // 问题：在同步块外检查
    synchronized (Client.messages){
        // ...
    }
}

// 原始Sender线程的等待逻辑
} else {
    try {
        Client.messages.wait(); // 问题：没有在while循环中检查条件
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
}

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;
    }
}

import java.util.concurrent.CopyOnWriteArrayList;

public class Sender extends Thread {
    private SMPPSession smppSession;
    private CopyOnWriteArrayList<String> messages; // 引用共享消息列表
    private volatile boolean running = true; // 控制线程生命周期

    public Sender(String name, SMPPSession smppSession, CopyOnWriteArrayList<String> 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;
            }
        }
    }
}

import java.util.concurrent.CopyOnWriteArrayList;

public class SessionProducer extends Thread {
    private SMPPSession smppSession;
    private CopyOnWriteArrayList<String> messages; // 引用共享消息列表，用于判断是否还有消息需要发送
    private volatile boolean running = true;

    public SessionProducer(String name, SMPPSession smppSession, CopyOnWriteArrayList<String> 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()