当前位置：首页 > news >正文

[Java实战]Spring Boot + Netty 实现 TCP 长连接客户端及 RESTful 请求转发（二十六）

news 来源：原创 2025/5/16 7:03:38

[Java实战]Spring Boot + Netty 实现 TCP 长连接客户端及 RESTful 请求转发（二十六）

在现代微服务架构中，经常需要在不同服务之间进行高效、可靠的通信。本文将介绍如何使用 Spring Boot 结合 Netty 实现一个 TCP 长连接客户端，并通过 RESTful 接口转发请求到后台 TCP 服务。这种架构在物联网、实时通信等领域非常常见。

一.架构设计

以下是系统的架构设计图：

HTTP Client：外部客户端通过 RESTful 接口发送请求。
Spring Boot Controller：接收 HTTP 请求并转发到 Netty 客户端服务。
Netty Client Service：管理 TCP 长连接，发送请求并接收响应。
TCP Connection Pool：管理多个 TCP 连接，提高性能和可靠性。
Backend TCP Server：后台 TCP 服务，处理实际的业务逻辑。

二.项目搭建与依赖配置

1. 创建项目并添加依赖

创建一个 Spring Boot 项目，并添加以下依赖：

<dependencies><!-- Spring Boot Web --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- Netty --><dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.94.Final</version></dependency><!-- Lombok --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency>
</dependencies>

2. 配置参数

在 application.yml 文件中配置 TCP 服务器的地址、端口以及其他参数：

tcp:server:host: 127.0.0.1port: 9000client:worker-threads: 4connect-timeout: 3000heartbeat-interval: 30000max-frame-length: 65535

三.核心组件实现

3.1 Netty 客户端启动器

NettyClientBootstrap 负责初始化 Netty 客户端并建立长连接：

@Component
@Slf4j
public class NettyClientBootstrap {@Value("${tcp.server.host}")private String host;@Value("${tcp.server.port}")private int port;private volatile Channel channel;private Bootstrap bootstrap;@PostConstructpublic void init() throws InterruptedException {EventLoopGroup workerGroup = new NioEventLoopGroup();bootstrap = new Bootstrap();bootstrap.group(workerGroup).channel(NioSocketChannel.class).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) {ChannelPipeline pipeline = ch.pipeline();// 协议处理pipeline.addLast(new LengthFieldPrepender(4));pipeline.addLast(new LengthFieldBasedFrameDecoder(65535, 0, 4, 0, 4));pipeline.addLast(new StringEncoder(StandardCharsets.UTF_8));pipeline.addLast(new StringDecoder(StandardCharsets.UTF_8));// 业务处理器pipeline.addLast(new ClientHandler());}});connect();}private void connect() throws InterruptedException {ChannelFuture future = bootstrap.connect(host, port).sync();channel = future.channel();channel.closeFuture().addListener(f -> {log.warn("Connection lost, reconnecting...");reconnect();});}private void reconnect() {bootstrap.connect(host, port).addListener((ChannelFutureListener) future -> {if (future.isSuccess()) {channel = future.channel();log.info("Reconnected successfully");} else {log.error("Reconnect failed, retrying in 5s");channel.eventLoop().schedule(this::reconnect, 5, TimeUnit.SECONDS);}});}public Channel getChannel() {if (!channel.isActive()) {throw new IllegalStateException("Channel is inactive");}return channel;}
}

3.2 业务处理器

ClientHandler 负责处理 TCP 通信中的请求和响应：

@Slf4j
public class ClientHandler extends SimpleChannelInboundHandler<String> {private static final ConcurrentMap<String, CompletableFuture<String>> pendingRequests = new ConcurrentHashMap<>();@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) {// 响应格式示例：REQ_ID|RESPONSE_DATAString[] parts = msg.split("\\|", 2);if (parts.length == 2) {CompletableFuture<String> future = pendingRequests.remove(parts[0]);if (future != null) {future.complete(parts[1]);}}}public static CompletableFuture<String> sendRequest(Channel channel, String message) {String reqId = UUID.randomUUID().toString();CompletableFuture<String> future = new CompletableFuture<>();pendingRequests.put(reqId, future);String protocolMsg = reqId + "|" + message;channel.writeAndFlush(protocolMsg).addListener(f -> {if (!f.isSuccess()) {future.completeExceptionally(f.cause());pendingRequests.remove(reqId);}});// 设置超时channel.eventLoop().schedule(() -> {if (pendingRequests.remove(reqId) != null) {future.completeExceptionally(new TimeoutException("Request timeout"));}}, 5, TimeUnit.SECONDS);return future;}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {log.error("Channel error", cause);ctx.close();}
}

3.3 服务层封装

TcpClientService 提供对外的服务接口，封装了 TCP 通信逻辑：

@Service
@RequiredArgsConstructor
public class TcpClientService {private final NettyClientBootstrap clientBootstrap;public String sendMessage(String message) throws Exception {Channel channel = clientBootstrap.getChannel();return ClientHandler.sendRequest(channel, message).get(5, TimeUnit.SECONDS);}
}

3.4 RESTful 接口层

TcpController 提供 RESTful 接口，接收外部请求并转发到 TCP 服务：

@RestController
@RequestMapping("/api")
@RequiredArgsConstructor
public class TcpController {private final TcpClientService tcpClientService;@PostMapping("/send")public ResponseEntity<?> sendCommand(@RequestBody String payload) {try {String response = tcpClientService.sendMessage(payload);return ResponseEntity.ok(response);} catch (TimeoutException e) {return ResponseEntity.status(HttpStatus.GATEWAY_TIMEOUT).body("Backend service timeout");} catch (Exception e) {return ResponseEntity.status(HttpStatus.BAD_GATEWAY).body("Communication error: " + e.getMessage());}}
}

3.5 心跳保活机制

HeartbeatScheduler 定期发送心跳消息，保持 TCP 连接活跃：

@Component
@RequiredArgsConstructor
public class HeartbeatScheduler {private final TcpClientService tcpClientService;@Scheduled(fixedRateString = "${tcp.client.heartbeat-interval}")public void heartbeat() {try {tcpClientService.sendMessage("HEARTBEAT");} catch (Exception e) {log.error("Heartbeat failed", e);}}
}

四.关键功能说明

1. 长连接管理

自动重连机制：断线后每 5 秒自动重试。
Channel 状态监控：实时监控连接状态，确保连接可用。
异常自动恢复：捕获异常并尝试恢复连接。

2. 协议设计

协议格式如下：

+----------------+-------------------+
| 32字节UUID     | 实际消息内容（UTF8）|
+----------------+-------------------+

3. 异步处理流程

异步处理流程如下：

五.测试方法

1. 启动模拟 TCP 服务端

使用 Python 快速搭建一个测试 TCP 服务端：

# 使用Python快速搭建测试服务
import socketsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(('0.0.0.0', 9000))
sock.listen(1)while True:conn, addr = sock.accept()while True:data = conn.recv(1024)if not data: breakreq_id, payload = data.decode().split('|', 1)conn.send(f"{req_id}|ECHO:{payload}".encode())conn.close()

2. 发送测试请求

通过 curl 发送测试请求：

curl -X POST -H "Content-Type: text/plain" \
-d "Hello Netty" \
http://localhost:8080/api/send

六.生产级优化建议

1. 连接池扩展

实现多 Channel 管理，提高性能和可靠性：

public class ConnectionPool {private final BlockingQueue<Channel> pool = new LinkedBlockingQueue<>(10);public Channel getChannel() {Channel channel = pool.poll();if (channel == null || !channel.isActive()) {channel = createNewChannel();}return channel;}private Channel createNewChannel() {// 创建新连接逻辑}
}

2. 监控指标

添加监控指标，便于实时监控系统状态：

@Bean
public MeterRegistryCustomizer<MeterRegistry> metrics() {return registry -> {Gauge.builder("tcp.active.connections", () -> clientBootstrap.getActiveCount()).register(registry);};
}

3. SSL 加密支持

为 TCP 连接添加 SSL 加密支持：

bootstrap.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) {SSLEngine engine = sslContext.newEngine(ch.alloc());pipeline.addLast(new SslHandler(engine));// 其他处理器...}
});

4. 流量控制

添加流量控制机制，防止服务过载：

// 在Handler中添加限流器
private final RateLimiter rateLimiter = RateLimiter.create(1000); // QPS=1000public static CompletableFuture<String> sendRequest(...) {if (!rateLimiter.tryAcquire()) {throw new RateLimitExceededException();}// 原有逻辑...
}