Java多线程环境下SMPP会话与消息发送的同步机制优化

// 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<String> messages = new CopyOnWriteArrayList<>(); // 共享消息列表
    public static void main(String[] args) {
        // ... 消息填充 ...
        SMPPSession smppSession = new SMPPSession();
        // ... 启动 SessionProducer 和 Sender 线程 ...
    }
}

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

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<String> messageQueue;
    private final AtomicBoolean running; // 用于控制线程生命周期

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