springboot使用netty做TCP客户端
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.1port: 15000timeout: 5000pool:maxTotal: 10maxIdle: 5minIdle: 24、 客户端示例
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@Datapublic 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);@Overrideprotected 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;}@Overridepublic 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);}}}@PreDestroypublic 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>() {@Overrideprotected 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());}@Overridepublic Channel create() throws Exception {return bootstrap.connect().sync().channel();}@Overridepublic PooledObject<Channel> wrap(Channel channel) {return new DefaultPooledObject<>(channel);}@Overridepublic boolean validateObject(PooledObject<Channel> p) {return p.getObject().isActive();}@Overridepublic void destroyObject(PooledObject<Channel> p) {p.getObject().close();}}static class LengthFieldDecoder extends ByteToMessageDecoder {@Overrideprotected 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> {@Overrideprotected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) {byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);out.writeIntLE(bytes.length); // 小端序写入长度字段out.writeBytes(bytes);}}
}