最新资讯

多内网服务器公网中转通信方案（Spring Boot 2.7 + MyBatis Plus）

一、整体架构说明

1. 核心架构：采用「公网中转服务端 + 内网客户端（A/B/…）」的TCP Socket架构，公网服务器作为消息转发枢纽，内网客户端通过该枢纽实现相互通信
2. 关键约束：内网客户端不直连公网MySQL，仅操作本地MySQL（通过MyBatis Plus）
3. 核心特性：支持自动重连、消息确认（防丢失）、心跳检测（保连接）、多客户端扩展、实时请求响应
4. 消息流转：客户端A→公网中转→客户端B（B实时处理）→公网中转→客户端A

二、统一依赖（服务端 + 客户端）

<dependencies>
    
    <dependency>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starterartifactId>
        <version>2.7.18version>
    dependency>

    
    <dependency>
        <groupId>com.alibabagroupId>
        <artifactId>fastjsonartifactId>
        <version>2.0.25version>
    dependency>

    
    <dependency>
        <groupId>com.baomidougroupId>
        <artifactId>mybatis-plus-boot-starterartifactId>
        <version>3.5.3version>
    dependency>

    
    <dependency>
        <groupId>mysqlgroupId>
        <artifactId>mysql-connector-javaartifactId>
        <version>8.0.30version>
        <scope>runtimescope>
    dependency>

    
    <dependency>
        <groupId>org.projectlombokgroupId>
        <artifactId>lombokartifactId>
        <optional>trueoptional>
    dependency>
dependencies>

三、通用核心组件（服务端 + 客户端共用）

1. 自定义消息实体（统一消息格式）

import lombok.Data;
import java.util.Date;

/**
 * 统一消息体（用于传输、本地存储）
 */
@Data
public class Message {
    /** 唯一消息ID（用于消息确认） */
    private String msgId;
    /** 发送方客户端标识（如client_a、client_b） */
    private String sender;
    /** 接收方客户端标识（如client_b、client_a，广播可填all） */
    private String receiver;
    /** 消息类型：REQUEST(请求)、RESPONSE(响应)、HEARTBEAT(心跳)、ACK(确认) */
    private String msgType;
    /** 消息内容（自定义JSON字符串） */
    private String content;
    /** 发送时间 */
    private Date sendTime;
    /** 确认ID（关联待确认的消息ID） */
    private String ackMsgId;
    /** 消息状态：SEND(已发送)、ACKED(已确认)、PROCESSED(已处理) */
    private String status;
}

2. 本地消息记录表（MySQL，内网客户端本地创建）

-- 本地消息记录表（每个内网客户端本地数据库执行）
CREATE TABLE `local_message` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '主键ID',
  `msg_id` varchar(64) NOT NULL COMMENT '消息唯一标识',
  `sender` varchar(32) NOT NULL COMMENT '发送方',
  `receiver` varchar(32) NOT NULL COMMENT '接收方',
  `msg_type` varchar(16) NOT NULL COMMENT '消息类型',
  `content` text COMMENT '消息内容',
  `send_time` datetime NOT NULL COMMENT '发送时间',
  `ack_msg_id` varchar(64) DEFAULT NULL COMMENT '确认消息ID',
  `status` varchar(16) NOT NULL COMMENT '消息状态',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_msg_id` (`msg_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='本地消息存储表';

3. MyBatis Plus 基础配置

（1）Message实体对应的Mapper

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import org.apache.ibatis.annotations.Mapper;

/**
 * 本地消息Mapper（内网客户端）
 */
@Mapper
public interface LocalMessageMapper extends BaseMapper<LocalMessage> {
    // 继承BaseMapper，无需额外编写CRUD方法
}

（2）LocalMessage 实体（对应数据库表）

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import java.util.Date;

@Data
@TableName("local_message")
public class LocalMessage {
    @TableId(type = IdType.AUTO)
    private Long id;
    private String msgId;
    private String sender;
    private String receiver;
    private String msgType;
    private String content;
    private Date sendTime;
    private String ackMsgId;
    private String status;
    private Date createTime;
    private Date updateTime;
}

四、公网中转服务端（核心：消息转发 + 连接管理）

1. 服务端配置（application.yml）

server:
  port: 8080 # Spring Boot 端口
socket:
  port: 8888 # Socket 监听端口（内网客户端连接端口）
  heartbeat-timeout: 30000 # 心跳超时时间（30秒，超过则断开连接）

2. 核心服务端代码（SocketServer）

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 公网中转Socket服务端（负责连接管理、消息转发、心跳检测）
 */
@Slf4j
@Component
public class SocketTransferServer implements ApplicationRunner {
    @Value("${socket.port}")
    private int socketPort;
    @Value("${socket.heartbeat-timeout}")
    private long heartbeatTimeout;

    // 客户端连接池：key=客户端标识，value=Socket连接（线程安全，支持多客户端）
    private static final Map<String, SocketClientHolder> CLIENT_MAP = new ConcurrentHashMap<>();
    // 定时任务池（心跳检测）
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    /**
     * Spring Boot启动后，启动Socket服务端
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 启动心跳检测任务（每10秒检测一次）
        scheduler.scheduleAtFixedRate(this::checkHeartbeat, 10, 10, TimeUnit.SECONDS);

        // 启动Socket服务端监听
        try (ServerSocket serverSocket = new ServerSocket(socketPort)) {
            log.info("公网中转服务端启动成功，监听端口：{}", socketPort);
            while (true) {
                // 阻塞等待客户端连接
                Socket clientSocket = serverSocket.accept();
                log.info("新客户端连接：{}", clientSocket.getInetAddress().getHostAddress());
                // 启动独立线程处理客户端消息
                new Thread(() -> handleClient(clientSocket)).start();
            }
        } catch (IOException e) {
            log.error("Socket服务端启动失败", e);
        }
    }

    /**
     * 处理单个客户端的消息读取与转发
     */
    private void handleClient(Socket clientSocket) {
        BufferedReader reader = null;
        try {
            reader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream(), StandardCharsets.UTF_8));
            String jsonStr;
            // 持续读取客户端消息
            while ((jsonStr = reader.readLine()) != null) {
                Message message = JSON.parseObject(jsonStr, Message.class);
                log.info("收到客户端消息：{}", jsonStr);

                // 1. 心跳消息：更新客户端最后活跃时间
                if ("HEARTBEAT".equals(message.getMsgType())) {
                    SocketClientHolder holder = CLIENT_MAP.get(message.getSender());
                    if (holder != null) {
                        holder.setLastActiveTime(System.currentTimeMillis());
                    }
                    continue;
                }

                // 2. 首次连接：注册客户端标识
                if (!CLIENT_MAP.containsKey(message.getSender())) {
                    CLIENT_MAP.put(message.getSender(), new SocketClientHolder(clientSocket, System.currentTimeMillis()));
                    log.info("客户端注册成功：{}", message.getSender());
                }

                // 3. 普通消息（REQUEST/RESPONSE/ACK）：转发给目标客户端
                SocketClientHolder targetHolder = CLIENT_MAP.get(message.getReceiver());
                if (targetHolder != null) {
                    OutputStreamWriter writer = new OutputStreamWriter(targetHolder.getSocket().getOutputStream(), StandardCharsets.UTF_8);
                    writer.write(jsonStr + "
");
                    writer.flush();
                    log.info("消息转发成功：{} -> {}", message.getSender(), message.getReceiver());
                } else {
                    log.warn("目标客户端未在线：{}", message.getReceiver());
                }
            }
        } catch (IOException e) {
            log.error("客户端连接异常", e);
        } finally {
            // 移除失效连接
            CLIENT_MAP.entrySet().removeIf(entry -> entry.getValue().getSocket() == clientSocket);
            try {
                if (reader != null) reader.close();
                clientSocket.close();
            } catch (IOException e) {
                log.error("关闭客户端连接失败", e);
            }
            log.info("客户端连接断开：{}", clientSocket.getInetAddress().getHostAddress());
        }
    }

    /**
     * 心跳检测：移除超时未活跃的客户端
     */
    private void checkHeartbeat() {
        long currentTime = System.currentTimeMillis();
        CLIENT_MAP.entrySet().removeIf(entry -> {
            long idleTime = currentTime - entry.getValue().getLastActiveTime();
            boolean timeout = idleTime > heartbeatTimeout;
            if (timeout) {
                log.warn("客户端心跳超时，移除连接：{}", entry.getKey());
                try {
                    entry.getValue().getSocket().close();
                } catch (IOException e) {
                    log.error("关闭超时客户端连接失败", e);
                }
            }
            return timeout;
        });
    }

    /**
     * 客户端连接持有者（存储Socket + 最后活跃时间）
     */
    static class SocketClientHolder {
        private Socket socket;
        private long lastActiveTime;

        public SocketClientHolder(Socket socket, long lastActiveTime) {
            this.socket = socket;
            this.lastActiveTime = lastActiveTime;
        }

        // getter/setter 省略（可通过lombok生成）
        public Socket getSocket() { return socket; }
        public void setSocket(Socket socket) { this.socket = socket; }
        public long getLastActiveTime() { return lastActiveTime; }
        public void setLastActiveTime(long lastActiveTime) { this.lastActiveTime = lastActiveTime; }
    }
}

五、内网客户端（A/B通用，仅修改配置标识）

1. 客户端配置（application.yml）

# 客户端唯一标识（client_a / client_b / client_c ...）
client:
  id: client_a
# 公网中转服务端配置
socket:
  server-ip: 1.2.3.4 # 公网服务器IP
  server-port: 8888 # 公网Socket端口
  reconnect-interval: 5000 # 重连间隔（5秒）
  heartbeat-interval: 10000 # 心跳发送间隔（10秒）
  msg-timeout: 20000 # 消息超时时间（20秒，未确认则重发）
  retry-count: 3 # 消息重发次数
# 本地MySQL配置（内网客户端本地数据库）
spring:
  datasource:
    url: jdbc:mysql://localhost:3306/local_db?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai
    username: root
    password: 123456
    driver-class-name: com.mysql.cj.jdbc.Driver
mybatis-plus:
  configuration:
    map-underscore-to-camel-case: true
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

2. 核心客户端代码（SocketClient + 自动重连 + 心跳 + 消息确认）

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 内网客户端（A/B通用）：自动重连 + 心跳 + 消息确认 + 实时消息处理
 */
@Slf4j
@Component
public class InnerSocketClient implements ApplicationRunner {
    @Value("${client.id}")
    private String clientId;
    @Value("${socket.server-ip}")
    private String serverIp;
    @Value("${socket.server-port}")
    private int serverPort;
    @Value("${socket.reconnect-interval}")
    private long reconnectInterval;
    @Value("${socket.heartbeat-interval}")
    private long heartbeatInterval;
    @Value("${socket.msg-timeout}")
    private long msgTimeout;
    @Value("${socket.retry-count}")
    private int retryCount;

    @Autowired
    private LocalMessageMapper localMessageMapper;

    // Socket连接
    private volatile Socket socket;
    // 待确认消息池：key=消息ID，value=消息详情（含重发次数）
    private final Map<String, PendingMsg> PENDING_MSG_MAP = new ConcurrentHashMap<>();
    // 定时任务池（重连 + 心跳 + 消息超时检测）
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3);
    // 消息发送锁（保证线程安全）
    private final Object sendLock = new Object();

    /**
     * Spring Boot启动后，初始化客户端
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 1. 启动自动重连任务
        scheduler.scheduleAtFixedRate(this::reconnect, 0, reconnectInterval, TimeUnit.MILLISECONDS);
        // 2. 启动心跳发送任务
        scheduler.scheduleAtFixedRate(this::sendHeartbeat, heartbeatInterval, heartbeatInterval, TimeUnit.MILLISECONDS);
        // 3. 启动消息超时检测任务（重发超时未确认消息）
        scheduler.scheduleAtFixedRate(this::checkMsgTimeout, 5, 5, TimeUnit.SECONDS);
    }

    /**
     * 自动重连：连接断开时自动重试
     */
    private void reconnect() {
        if (socket != null && socket.isConnected() && !socket.isClosed()) {
            return;
        }
        try {
            log.info("尝试连接公网服务端：{}:{}", serverIp, serverPort);
            Socket newSocket = new Socket(serverIp, serverPort);
            this.socket = newSocket;
            log.info("公网服务端连接成功！");
            // 连接成功后，发送注册消息（携带客户端标识）
            sendRegisterMsg();
            // 启动消息监听线程（实时接收消息）
            new Thread(this::listenMsg).start();
        } catch (IOException e) {
            log.error("连接公网服务端失败，{}毫秒后重试", reconnectInterval, e);
        }
    }

    /**
     * 发送注册消息（首次连接时）
     */
    private void sendRegisterMsg() {
        Message msg = new Message();
        msg.setMsgId(UUID.randomUUID().toString());
        msg.setSender(clientId);
        msg.setReceiver("server");
        msg.setMsgType("REGISTER");
        msg.setSendTime(new Date());
        msg.setStatus("SEND");
        sendMsg(msg);
    }

    /**
     * 发送心跳消息
     */
    private void sendHeartbeat() {
        if (socket == null || socket.isClosed()) {
            return;
        }
        Message msg = new Message();
        msg.setMsgId(UUID.randomUUID().toString());
        msg.setSender(clientId);
        msg.setReceiver("server");
        msg.setMsgType("HEARTBEAT");
        msg.setSendTime(new Date());
        msg.setStatus("SEND");
        sendMsg(msg);
    }

    /**
     * 发送自定义消息（对外提供的核心方法）
     * @param receiver 目标客户端标识（如client_b）
     * @param content  自定义消息内容
     * @return 消息ID
     */
    public String sendCustomMsg(String receiver, String content) {
        Message msg = new Message();
        String msgId = UUID.randomUUID().toString();
        msg.setMsgId(msgId);
        msg.setSender(clientId);
        msg.setReceiver(receiver);
        msg.setMsgType("REQUEST");
        msg.setContent(content);
        msg.setSendTime(new Date());
        msg.setStatus("SEND");
        // 1. 保存到本地数据库
        saveLocalMsg(msg);
        // 2. 发送消息并加入待确认池
        sendMsg(msg);
        PENDING_MSG_MAP.put(msgId, new PendingMsg(msg, 0, System.currentTimeMillis()));
        return msgId;
    }

    /**
     * 底层消息发送实现
     */
    private void sendMsg(Message msg) {
        synchronized (sendLock) {
            if (socket == null || socket.isClosed()) {
                log.error("Socket连接未建立，消息发送失败：{}", JSON.toJSONString(msg));
                return;
            }
            try {
                OutputStreamWriter writer = new OutputStreamWriter(socket.getOutputStream(), StandardCharsets.UTF_8);
                writer.write(JSON.toJSONString(msg) + "
");
                writer.flush();
                log.info("消息发送成功：{}", JSON.toJSONString(msg));
            } catch (IOException e) {
                log.error("消息发送失败，触发重连", e);
                // 发送失败，关闭Socket触发重连
                closeSocket();
            }
        }
    }

    /**
     * 实时监听公网服务端转发的消息
     */
    private void listenMsg() {
        BufferedReader reader = null;
        try {
            reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
            String jsonStr;
            // 持续监听消息（实时处理）
            while ((jsonStr = reader.readLine()) != null) {
                Message msg = JSON.parseObject(jsonStr, Message.class);
                log.info("收到转发消息：{}", jsonStr);
                // 1. 消息确认：移除待确认池并更新本地消息状态
                if ("ACK".equals(msg.getMsgType())) {
                    PENDING_MSG_MAP.remove(msg.getAckMsgId());
                    updateLocalMsgStatus(msg.getAckMsgId(), "ACKED");
                    continue;
                }
                // 2. 响应消息：业务处理（可自定义）
                if ("RESPONSE".equals(msg.getMsgType())) {
                    handleResponseMsg(msg);
                    // 发送确认消息
                    sendAckMsg(msg.getMsgId(), msg.getSender());
                    continue;
                }
                // 3. 请求消息：实时处理并返回响应
                if ("REQUEST".equals(msg.getMsgType())) {
                    handleRequestMsg(msg);
                    continue;
                }
            }
        } catch (IOException e) {
            log.error("消息监听异常，触发重连", e);
            closeSocket();
        } finally {
            try {
                if (reader != null) reader.close();
            } catch (IOException e) {
                log.error("关闭输入流失败", e);
            }
        }
    }

    /**
     * 处理请求消息（实时业务处理，可自定义）
     */
    private void handleRequestMsg(Message requestMsg) {
        try {
            // 1. 先发送确认消息（防止消息丢失）
            sendAckMsg(requestMsg.getMsgId(), requestMsg.getSender());
            // 2. 实时处理业务逻辑（此处为示例，可自定义）
            log.info("实时处理请求消息：{}", requestMsg.getContent());
            String responseContent = "处理完成：" + requestMsg.getContent() + "（来自" + clientId + "）";
            // 3. 发送响应消息
            Message responseMsg = new Message();
            responseMsg.setMsgId(UUID.randomUUID().toString());
            responseMsg.setSender(clientId);
            responseMsg.setReceiver(requestMsg.getSender());
            responseMsg.setMsgType("RESPONSE");
            responseMsg.setContent(responseContent);
            responseMsg.setSendTime(new Date());
            responseMsg.setStatus("SEND");
            // 保存响应消息到本地
            saveLocalMsg(responseMsg);
            // 发送响应消息
            sendMsg(responseMsg);
            // 更新请求消息本地状态
            updateLocalMsgStatus(requestMsg.getMsgId(), "PROCESSED");
        } catch (Exception e) {
            log.error("处理请求消息失败", e);
        }
    }

    /**
     * 处理响应消息（可自定义业务逻辑）
     */
    private void handleResponseMsg(Message responseMsg) {
        log.info("收到响应消息：{} -> {}，内容：{}",
                responseMsg.getSender(), clientId, responseMsg.getContent());
        // 业务处理：如更新业务状态、通知上层服务等
        updateLocalMsgStatus(responseMsg.getMsgId(), "PROCESSED");
    }

    /**
     * 发送消息确认（ACK）
     */
    private void sendAckMsg(String ackMsgId, String receiver) {
        Message ackMsg = new Message();
        ackMsg.setMsgId(UUID.randomUUID().toString());
        ackMsg.setSender(clientId);
        ackMsg.setReceiver(receiver);
        ackMsg.setMsgType("ACK");
        ackMsg.setAckMsgId(ackMsgId);
        ackMsg.setSendTime(new Date());
        ackMsg.setStatus("SEND");
        sendMsg(ackMsg);
    }

    /**
     * 检查消息超时：重发超时未确认的消息
     */
    private void checkMsgTimeout() {
        long currentTime = System.currentTimeMillis();
        PENDING_MSG_MAP.entrySet().forEach(entry -> {
            PendingMsg pendingMsg = entry.getValue();
            // 超时判断
            if (currentTime - pendingMsg.getSendTime() > msgTimeout) {
                // 重发次数判断
                if (pendingMsg.getRetryCount() < retryCount) {
                    log.info("消息超时，开始重发（第{}次）：{}",
                            pendingMsg.getRetryCount() + 1, entry.getKey());
                    // 重发消息
                    sendMsg(pendingMsg.getMessage());
                    // 更新重发次数和发送时间
                    pendingMsg.setRetryCount(pendingMsg.getRetryCount() + 1);
                    pendingMsg.setSendTime(System.currentTimeMillis());
                } else {
                    log.error("消息重发次数耗尽，发送失败：{}", entry.getKey());
                    // 移除待确认池，更新本地状态
                    PENDING_MSG_MAP.remove(entry.getKey());
                    updateLocalMsgStatus(entry.getKey(), "FAILED");
                }
            }
        });
    }

    /**
     * 保存消息到本地数据库
     */
    private void saveLocalMsg(Message msg) {
        LocalMessage localMessage = new LocalMessage();
        localMessage.setMsgId(msg.getMsgId());
        localMessage.setSender(msg.getSender());
        localMessage.setReceiver(msg.getReceiver());
        localMessage.setMsgType(msg.getMsgType());
        localMessage.setContent(msg.getContent());
        localMessage.setSendTime(msg.getSendTime());
        localMessage.setAckMsgId(msg.getAckMsgId());
        localMessage.setStatus(msg.getStatus());
        localMessageMapper.insert(localMessage);
    }

    /**
     * 更新本地消息状态
     */
    private void updateLocalMsgStatus(String msgId, String status) {
        LambdaUpdateWrapper<LocalMessage> updateWrapper = new LambdaUpdateWrapper<>();
        updateWrapper.eq(LocalMessage::getMsgId, msgId)
                     .set(LocalMessage::getStatus, status);
        localMessageMapper.update(null, updateWrapper);
    }

    /**
     * 关闭Socket连接
     */
    private void closeSocket() {
        if (socket != null && !socket.isClosed()) {
            try {
                socket.close();
            } catch (IOException e) {
                log.error("关闭Socket失败", e);
            }
        }
    }

    /**
     * 待确认消息持有者
     */
    static class PendingMsg {
        private Message message;
        private int retryCount;
        private long sendTime;

        public PendingMsg(Message message, int retryCount, long sendTime) {
            this.message = message;
            this.retryCount = retryCount;
            this.sendTime = sendTime;
        }

        // getter/setter 省略
        public Message getMessage() { return message; }
        public void setMessage(Message message) { this.message = message; }
        public int getRetryCount() { return retryCount; }
        public void setRetryCount(int retryCount) { this.retryCount = retryCount; }
        public long getSendTime() { return sendTime; }
        public void setSendTime(long sendTime) { this.sendTime = sendTime; }
    }
}

3. 客户端测试代码（发送消息示例）

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

/**
 * 客户端测试：启动后发送测试消息
 */
@Component
public class ClientTestRunner implements CommandLineRunner {
    @Autowired
    private InnerSocketClient innerSocketClient;

    @Override
    public void run(String... args) throws Exception {
        // 延迟5秒（确保连接建立成功）
        Thread.sleep(5000);
        // 发送测试消息给client_b（客户端A配置）
        // 若为客户端B，可发送给client_a
        String msgId = innerSocketClient.sendCustomMsg("client_b", "这是来自client_a的测试请求");
        log.info("发送测试消息，消息ID：{}", msgId);
    }
}

六、核心特性说明

1. 多客户端扩展：仅需修改application.yml中的client.id（如client_c、client_d），即可实现无限扩展
2. 自动重连：客户端检测到Socket断开（发送/监听异常），会每隔5秒自动重试连接，直到成功
3. 消息确认：发送方发送消息后加入待确认池，接收方收到消息后立即返回ACK，发送方超时未收到则自动重发（最多3次）
4. 心跳检测：客户端每10秒发送心跳，服务端30秒未收到心跳则移除连接，客户端检测到连接异常则触发重连
5. 实时处理：客户端启动独立线程持续监听消息，收到请求后立即处理并返回响应，无延迟
6. 消息持久化：所有消息均保存到内网客户端本地MySQL（通过MyBatis Plus），可追溯、防丢失

七、部署与使用步骤

1. 公网服务器部署：启动公网中转服务端（Spring Boot项目）
2. 内网客户端A部署：
   • 本地创建MySQL数据库并执行local_message表SQL
   • 修改application.yml：client.id=client_a、公网IP、本地MySQL配置
   • 启动客户端A
3. 内网客户端B部署：
   • 本地创建MySQL数据库并执行local_message表SQL
   • 修改application.yml：client.id=client_b、公网IP、本地MySQL配置
   • 启动客户端B
4. 自动通信：客户端A可发送消息给client_b，客户端B实时处理并返回响应，反之亦然