Java多线程并发消息发送与会话管理教程

// SMPPSession 类 (共享资源)
public class SMPPSession {
    private boolean bind; // 会话绑定状态
    public int sendMessage(String msg) { /* ... */ }
    public void reBind() { /* ... */ this.bind = true; }
    public 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); // 移除消息
                    // ... 发送消息
                    Client.messages.notifyAll(); // 问题点2: 在此调用notifyAll可能不合适
                } else {
                    try {
                        Client.messages.wait(); // 等待会话恢复
                    } catch (InterruptedException e) { /* ... */ }
                }
            }
        }
    }
}

// SessionProducer 线程
public class SessionProducer extends Thread {
    private SMPPSession smppSession;
    // ... 构造函数
    @Override
    public void run() {
        while (!Client.messages.isEmpty()){ // 问题点3: 生产者不应依赖消息列表是否为空
            synchronized (Client.messages){ // 锁定共享消息列表
                if (!smppSession.isBind()){
                    smppSession.reBind(); // 重新绑定
                    Client.messages.notifyAll(); // 唤醒所有等待线程
                } else{
                    try {
                        Client.messages.wait(); // 等待会话断开
                    } catch (InterruptedException e) { /* ... */ }
                }
            }
        }
    }
}

// Client 类 (主程序)
public class Client {
    public static final List<String> messages = new CopyOnWriteArrayList<>(); // 共享消息列表
    // ... main 方法启动线程
}

public class SMPPSession {
    private volatile boolean bind = false; // 初始化为false，并声明为volatile
    private static final Random idGenerator = new Random();

    public int sendMessage(String msg){
       try{
           Thread.sleep(1000L);
           System.out.println("Sending message: " + msg);
           return Math.abs(idGenerator.nextInt());
       } catch (InterruptedException e){
           Thread.currentThread().interrupt(); // 重新设置中断标志
           e.printStackTrace();
       }
        return -1;
    }

    public void reBind(){
        try{
            System.out.println("Rebinding...");
            Thread.sleep(1000L);
            this.bind = true; // 更新绑定状态
            System.out.println("Session established!");
        } catch (InterruptedException e){
            Thread.currentThread().interrupt(); // 重新设置中断标志
            e.printStackTrace();
        }
    }

    public boolean isBind(){
        return this.bind;
    }
}

public class Sender extends Thread{

    private SMPPSession smppSession;

    public Sender(String name, SMPPSession smppSession){
        this.setName(name);
        this.smppSession = smppSession;
    }

    @Override
    public void run(){
        while (true){ // 持续运行，直到所有消息发送完毕或被中断
            String msgToSend = null;
            synchronized (Client.messages){ // 锁定共享消息列表
                // 条件检查：会话未绑定 或 消息列表为空
                while (!smppSession.isBind() || Client.messages.isEmpty()){
                    // 如果消息列表为空且会话已绑定，说明所有消息已发送，此发送者线程可以退出
                    if (Client.messages.isEmpty() && smppSession.isBind()){
                        System.out.println(getName() + " finished sending all messages. Terminating.");
                        return; // 退出线程
                    }
                    try {
                        System.out.println(getName() + " waiting. Bind status: " + smppSession.isBind() + ", Messages empty: " + Client.messages.isEmpty());
                        Client.messages.wait(); // 等待条件满足
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt(); // 重新设置中断标志
                        System.out.println(getName() + " interrupted while waiting. Terminating.");
                        return; // 退出线程
                    }
                }
                // 条件满足：会话已绑定且有消息可发送
                msgToSend = Client.messages.remove(0); // 在同步块内安全移除消息
                // 注意：这里不需要notifyAll，因为Sender线程移除消息不会改变SessionProducer的等待条件，
                // 也不会让其他Sender线程立即从等待中恢复（除非它们也等待消息出现，但这里它们等待的是isBind和消息非空）
            }

            // 在同步块外部执行耗时的消息发送操作
            final int msgId = smppSession.sendMessage(msgToSend);
            System.out.println(getName() + " sent msg and received msgId: " + msgId);
        }
    }
}

public class SessionProducer extends Thread{

    private SMPPSession smppSession;

    public SessionProducer(String name, SMPPSession smppSession){
        this.setName(name);
        this.smppSession = smppSession;
    }

    @Override
    public void run(){
        while (true){ // 持续运行，监控并维护会话状态
            synchronized (Client.messages){ // 锁定共享消息列表
                // 条件检查：如果会话已绑定，则等待
                while (smppSession.isBind()){
                    // 如果所有消息都已发送，且会话已绑定，SessionProducer可以等待新的消息出现或等待终止信号
                    // 暂时让它等待，因为它的核心职责是确保会话可用。
                    if (Client.messages.isEmpty()) {
                        System.out.println(getName() + " session is bound and messages are empty. Waiting for potential new messages or session break.");
                    } else {
                        System.out.println(getName() +