1、服务端文档说明

## 1. 概述

本文档描述了Socket模拟器的通信协议实现细节，包括数据包格式、字节序、编码方式等信息。

## 2. 通信基础

### 2.1 连接方式
- 协议类型：TCP
- 网络层：IPv4 (AddressFamily.InterNetwork)
- 传输方式：流式 (SocketType.Stream)
- 协议：TCP (ProtocolType.Tcp)

### 2.2 字节序
- 使用小端字节序（Little-Endian）
- 使用 `BitConverter` 进行字节序列的转换
- 长度字段采用4字节整数表示

## 3. 数据包格式

### 3.1 基本结构
```
+----------------+------------------+
|  Length (4B)   |   Payload       |
+----------------+------------------+
```

- Length: 4字节整数，表示Payload的长度
- Payload: UTF-8编码的消息内容

### 3.2 字段说明
1. Length字段
   - 大小：4字节
   - 类型：Int32
   - 字节序：小端序
   - 说明：表示后续Payload的字节长度

2. Payload字段
   - 编码：UTF-8
   - 长度：可变，由Length字段指定
   - 内容：实际传输的消息数据

## 4. 消息处理流程

### 4.1 发送流程
1. 将消息字符串转换为UTF-8字节数组
2. 计算消息字节数组长度
3. 将长度转换为4字节数组（小端序）
4. 组合长度字段和消息内容
5. 发送完整数据包

示例代码：
```csharp
byte[] bytes = Encoding.UTF8.GetBytes(sendMsg);   
byte[] xLenAry = BitConverter.GetBytes(bytes.Length);
byte[] sendData = new byte[bytes.Length + 4];
xLenAry.CopyTo(sendData, 0);
bytes.CopyTo(sendData, 4);
socket.Send(sendData);
```

### 4.2 接收流程
1. 接收数据到缓冲区（缓冲区大小1MB）
2. 读取前4字节获取消息长度
3. 根据长度读取后续消息内容
4. 将字节数组转换为UTF-8字符串

示例代码：
```csharp
byte[] arrServerRecMsg = new byte[1024 * 1024];
int length = socket.Receive(arrServerRecMsg);
byte[] lenstr = arrServerRecMsg.Skip(0).Take(4).ToArray();
int len = BitConverter.ToInt32(lenstr, 0);
string strSRecMsg = Encoding.UTF8.GetString(arrServerRecMsg, 4, len);
```

## 5. 错误处理

### 5.1 连接断开检测
- 通过 `Socket.Connected` 属性检查连接状态
- 捕获异常处理连接断开情况
- 在连接断开时清理资源并通知UI

### 5.2 异常处理
- 捕获Socket异常并进行相应处理
- 在连接断开时关闭Socket
- 从连接池中移除断开的连接
- 更新UI显示连接状态

## 6. 缓冲区管理

### 6.1 接收缓冲区
- 大小：1MB (1024 * 1024 字节)
- 类型：字节数组
- 用途：临时存储接收到的数据

### 6.2 发送缓冲区
- 动态分配，根据消息长度创建
- 包含4字节长度头部和消息内容
- 一次性发送完整数据包

## 7. 注意事项

1. 字符编码统一使用UTF-8，支持中文等多语言字符
2. 发送消息时需要先发送长度信息
3. 接收消息时需要先解析长度字段
4. 所有网络操作都需要进行异常处理
5. 在连接断开时要及时清理资源

## 8. 性能考虑

1. 使用后台线程处理接收消息
2. 设置适当的缓冲区大小
3. 及时关闭不使用的连接
4. 避免频繁的字符串转换操作 

2、pom文件

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.5.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.86.Final</version>
        </dependency>

3、配置文件

tcp:
  client:
    host: 127.0.0.1
    port: 15000
    timeout: 5000
    pool:
      maxTotal: 10
      maxIdle: 5
      minIdle: 2

4、 客户端示例 

     4.1、配置类

package com.netty.client.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
@Data
@Configuration
@ConfigurationProperties(prefix = "tcp.client")
public class TcpClientConfig {
    private String host;
    private int port;
    private Pool pool;
    private int timeout;

    // Getters and Setters
    @Data
    public static class Pool {
        private int maxTotal;
        private int maxIdle;
        private int minIdle;

        // Getters and Setters
    }
}

4.2、ClientHandler

package com.netty.client.hander;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.concurrent.*;
@Slf4j
@Component
@ChannelHandler.Sharable
public class ClientHandler extends SimpleChannelInboundHandler<String> {
    private final ConcurrentMap<String, CompletableFuture<String>> pendingRequests = new ConcurrentHashMap<>();
    private static final ScheduledExecutorService TIMEOUT_EXECUTOR = Executors.newScheduledThreadPool(1);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        log.info("收到消息 ===> {}", msg);
        try{
            JSONObject jsonObject = JSON.parseObject(msg);
            JSONObject header = jsonObject.getJSONObject("Header");
            String correlationId = header.getString("TransactionID");
            CompletableFuture<String> future = pendingRequests.remove(correlationId);
            if (future != null) {
                future.complete(msg);
            }
        }catch (Exception e){
            log.info("channelRead0 has error ==>{}", Arrays.toString(e.getStackTrace()));
            log.info("channelRead0 message ==>{}",e.getMessage());
        }
    }

    public CompletableFuture<String> prepareResponse(String correlationId) {
        CompletableFuture<String> future = new CompletableFuture<>();
        ScheduledFuture<?> timeout = TIMEOUT_EXECUTOR.schedule(() -> {
            if (future.completeExceptionally(new TimeoutException())) {
                pendingRequests.remove(correlationId);
            }
        }, 5, TimeUnit.SECONDS);

        future.whenComplete((r, t) -> {
            timeout.cancel(true);
            pendingRequests.remove(correlationId);
        });

        pendingRequests.put(correlationId, future);

        return future;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        ctx.close();
    }

}

4.3、client

package com.netty.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.netty.client.config.TcpClientConfig;
import com.netty.client.hander.ClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@Slf4j
@Component
public class NettyTcpClient {
    private final GenericObjectPool<Channel> connectionPool;
    private final ClientHandler clientHandler;
    private final TcpClientConfig config;

    public NettyTcpClient(TcpClientConfig config) {
        this.config = config;
        this.clientHandler = new ClientHandler();
        this.connectionPool = new GenericObjectPool<>(
                new ChannelFactory(config, clientHandler),
                buildPoolConfig(config.getPool())
        );
    }

    private GenericObjectPoolConfig<Channel> buildPoolConfig(TcpClientConfig.Pool poolConfig) {
        GenericObjectPoolConfig<Channel> config = new GenericObjectPoolConfig<>();
        config.setMaxTotal(poolConfig.getMaxTotal());
        config.setMaxIdle(poolConfig.getMaxIdle());
        config.setMinIdle(poolConfig.getMinIdle());
        return config;
    }

    public String sendSync(String message) throws Exception {
        Channel channel = null;
        try {
            channel = connectionPool.borrowObject();
            JSONObject jsonObject = JSON.parseObject(message);
            JSONObject header = jsonObject.getJSONObject("header");
            String correlationId = header.getString("transactionID");
            // String correlationId = UUID.randomUUID().toString();
            CompletableFuture<String> future = clientHandler.prepareResponse(correlationId);
            channel.writeAndFlush(message).sync();
            return future.get(config.getTimeout(), TimeUnit.MILLISECONDS);
        }
        finally {
            if (channel != null) {
                connectionPool.returnObject(channel);
            }
        }
    }

    @PreDestroy
    public void shutdown() {
        connectionPool.close();
    }

    private static class ChannelFactory extends BasePooledObjectFactory<Channel> {
        private final Bootstrap bootstrap;
        private final ClientHandler handler;

        public ChannelFactory(TcpClientConfig config, ClientHandler handler) {
            this.handler = handler;
            this.bootstrap = new Bootstrap()
                    .group(new NioEventLoopGroup())
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 解码器
                            pipeline.addLast(new LengthFieldDecoder());
                            // 编码器
                            pipeline.addLast(new LengthFieldEncoder());
                            // 业务处理器
                            pipeline.addLast(handler);
                        }
                    })
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                    .remoteAddress(config.getHost(), config.getPort());
        }

        @Override
        public Channel create() throws Exception {
            return bootstrap.connect().sync().channel();
        }

        @Override
        public PooledObject<Channel> wrap(Channel channel) {
            return new DefaultPooledObject<>(channel);
        }

        @Override
        public boolean validateObject(PooledObject<Channel> p) {
            return p.getObject().isActive();
        }

        @Override
        public void destroyObject(PooledObject<Channel> p) {
            p.getObject().close();
        }
    }
    static class LengthFieldDecoder extends ByteToMessageDecoder {
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
            if (in.readableBytes() < 4) return;

            in.markReaderIndex();
            int length = in.readIntLE();  // 小端序读取长度字段
            if (in.readableBytes() < length) {
                in.resetReaderIndex();
                return;
            }

            ByteBuf payload = in.readBytes(length);
            out.add(payload.toString(StandardCharsets.UTF_8));
        }
    }

    static class LengthFieldEncoder extends MessageToByteEncoder<String> {
        @Override
        protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) {
            byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
            out.writeIntLE(bytes.length);  // 小端序写入长度字段
            out.writeBytes(bytes);
        }
    }
}

修改发送消息和接收消息方法

package com.unisic.web.schaeffler.client;


import com.unisic.web.schaeffler.config.TcpClientConfig;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@Data
@Component
public class NettyClient {
    private TcpClientConfig config;
    private Channel channel;
    private EventLoopGroup workerGroup;
    private final ReentrantLock sendLock = new ReentrantLock();
    private CompletableFuture<String> currentFuture;
    private final Condition responseReceived = sendLock.newCondition();
    public NettyClient() {

    }
    public NettyClient(TcpClientConfig config) {
        this.config = config;
    }

    public void start() throws InterruptedException {
        workerGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new LengthFieldDecoder());
                            pipeline.addLast(new LengthFieldEncoder());
                            pipeline.addLast(new ClientHandler());
                        }
                    });

            ChannelFuture future = bootstrap.connect(config.getHost(), config.getPort()).sync();
            channel = future.channel();
            channel.closeFuture().addListener(f -> workerGroup.shutdownGracefully());
        } catch (InterruptedException e) {
            workerGroup.shutdownGracefully();
            throw e;
        }
    }
    public String sendSync(String message) throws InterruptedException, TimeoutException {
        log.info("mes Client send message ==> {}", message);

        sendLock.lock();
        try {
            if (currentFuture != null) {
                throw new IllegalStateException("Another request is in progress");
            }

            currentFuture = new CompletableFuture<>();
            final long startTime = System.currentTimeMillis();
            final long timeoutMs = TimeUnit.SECONDS.toMillis(config.getTimeout());

            // 发送消息
            channel.writeAndFlush(message).addListener(f -> {
                if (!f.isSuccess() && currentFuture != null) {
                    completeFutureExceptionally(f.cause());
                }
            });

            // 使用条件变量等待响应
            while (currentFuture != null && !currentFuture.isDone()) {
                long elapsed = System.currentTimeMillis() - startTime;
                long remaining = timeoutMs - elapsed;

                if (remaining <= 0) {
                    throw new TimeoutException("Request timed out after " + config.getTimeout() + " seconds");
                }

                // 等待剩余时间或收到信号
                try {
                    responseReceived.await(remaining, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw e;
                }
            }

            if (currentFuture != null) {
                return currentFuture.getNow(null);
            } else {
                throw new RuntimeException("Request was cancelled");
            }
        } finally {
            currentFuture = null;
            sendLock.unlock();
        }
    }

    // 修改点2：安全的 future 完成方法
    private void completeFuture(String response) {
        sendLock.lock();
        try {
            if (currentFuture != null) {
                currentFuture.complete(response);
                responseReceived.signalAll();
            }
        } finally {
            sendLock.unlock();
        }
    }

    private void completeFutureExceptionally(Throwable cause) {
        sendLock.lock();
        try {
            if (currentFuture != null) {
                currentFuture.completeExceptionally(cause);
                responseReceived.signalAll();
            }
        } finally {
            sendLock.unlock();
        }
    }

    public void shutdown() {
        if (channel != null) {
            channel.close();
        }
        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
        }
    }
    // 修改点4：解码器优化
    static class LengthFieldDecoder extends LengthFieldBasedFrameDecoder {
        public LengthFieldDecoder() {
            // 使用更合理的参数设置
            super(65535, 0, 4, 0, 4);
        }

        @Override
        protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
            ByteBuf frame = (ByteBuf) super.decode(ctx, in);
            if (frame == null) return null;

            try {
                String rawMessage = frame.toString(StandardCharsets.UTF_8);
                log.info("mes Client receive message ==> {}", rawMessage);
                return rawMessage;
            } finally {
                frame.release(); // 重要：释放ByteBuf
            }
        }
    }

    static class LengthFieldEncoder extends MessageToByteEncoder<String> {
        @Override
        protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) {
            // 1. 添加4字节长度头（大端序）
            byte[] content = msg.getBytes(StandardCharsets.UTF_8);
            out.writeInt(content.length);  // Big Endian
            // 2. 写入消息内容
            out.writeBytes(content);
        }
    }

    // 修改点3：更新处理器
    class ClientHandler extends SimpleChannelInboundHandler<String> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) {
            log.info("mes Client Received response: {}", msg);
            completeFuture(msg);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            log.error("mes Client Channel error", cause);
            completeFutureExceptionally(cause);
            ctx.close();
        }
    }
}