Java多线程会话管理与并发消息发送最佳实践

import java.util.Random;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class SMPPSession {

    private volatile boolean bind; // 使用volatile确保可见性
    private static final Random idGenerator = new Random();

    // 用于保护bind状态和协调线程的锁
    private final ReentrantLock sessionLock = new ReentrantLock();
    // 条件变量：当会话可用时通知等待的线程
    private final Condition sessionAvailable = sessionLock.newCondition();

    public SMPPSession() {
        this.bind = false; // 初始状态为未绑定
    }

    public int sendMessage(String msg) {
        sessionLock.lock(); // 保护会话状态和发送操作
        try {
            // 循环检查会话是否绑定，防止虚假唤醒
            while (!bind) {
                System.out.println(Thread.currentThread().getName() + "：会话未绑定，等待重连...");
                sessionAvailable.await(); // 等待会话可用
            }
            // 会话已绑定，执行发送操作
            Thread.sleep(100L); // 模拟发送延迟
            System.out.println("发送消息: " + msg);
            return Math.abs(idGenerator.nextInt());
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重新设置中断标志
            System.err.println("发送消息线程被中断: " + e.getMessage());
        } finally {
            sessionLock.unlock();
        }
        return -1;
    }

    public void reBind() {
        sessionLock.lock(); // 保护会话状态和重连操作
        try {
            if (bind) { // 如果已经绑定，则无需重连
                System.out.println("会话已绑定，无需重连。");
                return;
            }
            System.out.println(Thread.currentThread().getName() + "：正在重连会话...");
            Thread.sleep(2000L); // 模拟重连延迟
            this.bind = true;
            System.out.println(Thread.currentThread().getName() + "：会话已建立!");
            sessionAvailable.signalAll(); // 通知所有等待的线程会话已可用
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("重连线程被中断: " + e.getMessage());
        } finally {
            sessionLock.unlock();
        }
    }

    // 检查会话是否绑定，这个方法在同步块外调用时，只提供一个快照信息
    // 实际判断是否能发送消息应在sendMessage内部通过锁保护
    public boolean isBind() {
        return bind;
    }

    // 提供一个设置绑定状态的方法，供外部（例如模拟会话断开）调用
    public void setBind(boolean bind) {
        sessionLock.lock();
        try {
            this.bind = bind;
            if (bind) {
                sessionAvailable.signalAll(); // 如果设置为绑定，也通知一下
            }
        } finally {
            sessionLock.unlock();
        }
    }
}

import java.util.concurrent.BlockingQueue;

public class Sender extends Thread {

    private final SMPPSession smppSession;
    private final BlockingQueue<String> messageQueue; // 使用BlockingQueue

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

    @Override
    public void run() {
        try {
            while (true) { // 持续尝试发送消息
                String msg = messageQueue.take(); // 从队列中取出消息，如果队列为空则阻塞
                if (msg == null) { // 结束标志，如果队列设计有null作为结束
                    break;
                }
                int msgId = smppSession.sendMessage(msg);
                if (msgId != -1) {
                    System.out.println(Thread.currentThread().getName() + " 发送消息并收到ID: " + msgId);
                } else {
                    // 消息发送失败，可以考虑重新放入队列或记录日志
                    System.err.println(Thread.currentThread().getName() + " 消息发送失败: " + msg);
                    messageQueue.put(msg); // 重新放回队列，稍后重试
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println(Thread.currentThread().getName() + " 线程被中断.");
        }
    }
}

public class SessionProducer extends Thread {

    private final SMPPSession smppSession;

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

    @Override
    public void run() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                if (!smppSession.isBind()) { // 检查会话是否绑定
                    smppSession.reBind(); // 如果未绑定，则重连
                }
                // 即使会话已绑定，也等待一段时间再检查，避免忙循环
                Thread.sleep(5000); // 每隔5秒检查一次会话状态
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println(Thread.currentThread().getName() + " 线程被中断.");
        }
    }
}

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class Client {

    // 使用BlockingQueue作为消息队列
    public static final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();

    public static void main(String[] args) throws InterruptedException {

        // 填充消息
        messageQueue.put("msg1");
        messageQueue.put("msg2");
        messageQueue.put("msg3");
        messageQueue.put("msg4");
        messageQueue.put("msg5");
        messageQueue.put("msg6");
        messageQueue.put("msg7");
        messageQueue.put("msg8");
        messageQueue.put("msg9");
        messageQueue.put("msg10");


        SMPPSession smppSession = new SMPPSession();
        smppSession.setBind(false); // 初始设置为未绑定，模拟需要重连

        SessionProducer sessionProducer = new SessionProducer("SessionProducer1", smppSession);
        Sender sender1 = new Sender("Sender1", smppSession, messageQueue);
        Sender sender2 = new Sender("Sender2", smppSession, messageQueue);
        Sender sender3 = new Sender("Sender3", smppSession, messageQueue);
        Sender sender4 = new Sender("Sender4", smppSession, messageQueue);

        sessionProducer.start();
        sender1.start();
        sender2.start();
        sender3.start();
        sender4.start();

        // 模拟一段时间后停止所有线程
        Thread.sleep(20000); // 运行20秒

        // 优雅停机：中断所有线程
        sessionProducer.interrupt();
        sender1.interrupt();
        sender2.interrupt();
        sender3.interrupt();
        sender4.interrupt();

        System.out.println("所有线程已尝试停止。");
    }
}