springboot使用netty做TCP客户端

server/2025/3/31 5:12:31/

 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: 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@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);}}
}


http://www.ppmy.cn/server/179739.html

相关文章

VO、DTO、POJO、PO和DO 的区别

在 Java 开发中&#xff0c;VO、DTO、POJO、PO、DO 等概念经常被使用&#xff0c;它们的主要区别在于 用途 和 设计目的。 &#x1f525; 1. VO&#xff08;View Object&#xff09;—— 视图对象 目的&#xff1a; 用于前端展示&#xff0c;通常是后端返回给前端的数据格式。 …

Perl 环境安装指南

Perl 环境安装指南 引言 Perl是一种广泛使用的解释型、动态编程语言,以其强大的文本处理能力和灵活性著称。本文将为您详细介绍Perl环境的安装过程,包括系统要求、安装步骤以及注意事项。 系统要求 在安装Perl之前,请确保您的计算机满足以下基本要求: 操作系统:Window…

从单机到集群:Elasticsearch集群搭建指南

Elasticsearch是一个分布式搜索和分析引擎&#xff0c;广泛应用于日志分析、全文检索、实时数据分析等场景。在生产环境中&#xff0c;通常需要搭建多节点的Elasticsearch集群以提高系统的可用性、性能和容错能力。本文将详细介绍如何搭建一个Elasticsearch集群。 1 环境准备 …

【linux指令】一文掌握 Linux 基础命令(Linux 命令速查)

文章目录 命令速查表系统硬件用户登陆文件进程安装包文件权限安装源(编译)压缩/打包搜索网络文件传输磁盘使用情况目录遍历文件描述符输出重定向 前后台&&#xff08;终端关闭&#xff0c;程序也关闭&#xff09;nohup&#xff08;终端关闭&#xff0c;程序继续运行&#x…

Android UI 组件系列(三):ImageView 使用技巧与图像加载

引言 在 Android 开发中&#xff0c;图像展示是常见的需求之一&#xff0c;无论是在应用的界面中展示本地图片&#xff0c;还是从网络加载动态内容&#xff0c;ImageView 都是最基础的 UI 组件之一。它不仅能够显示各种类型的图像资源&#xff0c;还可以灵活地调整图像的显示方…

A Brief History: from GPT-1 to GPT-3

This is my reading notes of 《Developing Apps with GPT-4 and ChatGPT》. In this section, we will introduce the evolution of the OpenAI GPT medels from GPT-1 to GPT-4. GPT-1 In mid-2018, OpenAI published a paper titled “Improving Language Understanding …

阶段一:Java基础语法

目标&#xff1a;掌握Java的基本语法&#xff0c;理解变量、数据类型、运算符、控制结构等。 1. Java开发环境搭建 安装JDK配置环境变量编写第一个Java程序 代码示例&#xff1a; // HelloWorld.java public class HelloWorld { // 定义类名为 HelloWorldpublic static vo…

解决docker重启后包丢失的报错日常

项目是使用fastapi结合iotdb部署在阿里云服务器&#xff0c;今天修改了一个iotdb配置文件的问题&#xff0c;就一直显示500报错出现了&#xff0c;直接down了iotdb但是发现不仅是iotdb而是整个fastapi和iotdb的服务全部down完了&#xff0c;重启服务。 解决措施第一步 &#xf…