Spring Boot与Netty打造TCP服务端
- 前言
- 功能目标
- 项目实现
- maven坐标
- 构建自定义Handler
- ChannelInitializer实现
- server实现
前言
在物联网时代，设备之间的通信变得愈发重要。本文将带你踏上一场关于如何用Spring Boot和Netty搭建TCP服务端的冒险之旅。无论是智能家居、工业自动化还是其他物联网应用，构建一个稳健的通信桥梁将成为连接未来的关键。
功能目标
- 实现springboot+netty整合TCP服务端（基础）
- 实现消息回复功能
- 实现消息太长导致的粘包问题(比如发送一个base64的图片信息)
- 实现在自定义Handler中注入spring的bean
- 保证完成任务，哈哈哈哈哈
项目实现
maven坐标
<!-- netty 这里你也可以引入全部-->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-common</artifactId>
<version>4.1.79.Final</version>
</dependency>
构建自定义Handler
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
/**
* I/O数据读写处理类
*
* @author xiaobo
*/
@Slf4j
public class CarTcpNettyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter {
/**
* 从客户端收到新的数据时，这个方法会在收到消息时被调用
*
* @param ctx
* @param msg
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception, IOException {
// 这里是在前面的DelimiterBasedFrameDecoder转为了ByteBuf，验证是否是ByteBuf
if (msg instanceof ByteBuf) {
ByteBuf byteBuf = (ByteBuf) msg;
try {
String receivedData = byteBuf.toString(CharsetUtil.UTF_8);
// 接收完整数据
handleReceivedData(receivedData);
} finally {
// 释放 ByteBuf 占用的资源
byteBuf.release();
// 回复消息
ctx.writeAndFlush(Unpooled.copiedBuffer("收到over", CharsetUtil.UTF_8));
}
}
}
private void handleReceivedData(String receivedData) {
// 数据处理
// 这里如果想实现spring中bean的注入，可以用geBean的方式获取
}
/**
* 从客户端收到新的数据、读取完成时调用
*
* @param ctx
*/
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws IOException {
log.info("channelReadComplete");
ctx.flush();
}
/**
* 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
*
* @param ctx
* @param cause
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws IOException {
cause.printStackTrace();
ctx.close();// 抛出异常，断开与客户端的连接
}
/**
* 客户端与服务端第一次建立连接时 执行
*
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException {
super.channelActive(ctx);
ctx.channel().read();
InetSocketAddress socket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = socket.getAddress().getHostAddress();
// 此处不能使用ctx.close()，否则客户端始终无法与服务端建立连接
System.out.println("channelActive:" + clientIp + ctx.name());
// 这里是向客户端发送回应
ctx.writeAndFlush(Unpooled.copiedBuffer("收到over", CharsetUtil.UTF_8));
}
/**
* 客户端与服务端 断连时 执行
*
* @param ctx
* @throws Exception
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException {
super.channelInactive(ctx);
InetSocketAddress socket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = socket.getAddress().getHostAddress();
// 断开连接时，必须关闭，否则造成资源浪费，并发量很大情况下可能造成宕机
ctx.close();
log.info("channelInactive:{}", clientIp);
}
/**
* 服务端当read超时, 会调用这个方法
*
* @param ctx
* @param evt
* @throws Exception
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException {
super.userEventTriggered(ctx, evt);
InetSocketAddress socket = (InetSocketAddress) ctx.channel().remoteAddress();
String clientIp = socket.getAddress().getHostAddress();
ctx.close();// 超时时断开连接
log.info("userEventTriggered:" + clientIp);
}
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
log.info("channelRegistered");
}
@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
log.info("channelUnregistered");
}
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
log.info("channelWritabilityChanged");
}
}
ChannelInitializer实现
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
/**
* description: <h1>通道初始化</h1>
*
* @author bo
* @version 1.0
* @date 2024/2/27 16:13
*/
public class CarTcpNettyChannelInitializer<SocketChannel> extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel ch) throws Exception {
ByteBuf delemiter = Unpooled.buffer();
delemiter.writeBytes("$".getBytes());
// 这里就是解决数据过长问题，而且数据是以$结尾的
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(907200, true, true, delemiter));
// 自定义ChannelInboundHandlerAdapter
ch.pipeline().addLast(new CarTcpNettyChannelInboundHandlerAdapter());
}
}
server实现
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.AdaptiveRecvByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
/**
* description: <h1>netty创建的TCP</h1>
*
* @author bo
* @version 1.0
* @date 2024/2/27 16:25
*/
@Slf4j
public class CarTcpNettyServer {
public void bind(int port) throws Exception {
// 配置服务端的NIO线程组
// NioEventLoopGroup 是用来处理I/O操作的Reactor线程组
// bossGroup：用来接收进来的连接，workerGroup：用来处理已经被接收的连接,进行socketChannel的网络读写，
// bossGroup接收到连接后就会把连接信息注册到workerGroup
// workerGroup的EventLoopGroup默认的线程数是CPU核数的二倍
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
// netty 默认数据包传输大小为1024字节, 设置它可以自动调整下一次缓冲区建立时分配的空间大小，避免内存的浪费 最小 初始化 最大 (根据生产环境实际情况来定)
// 使用对象池，重用缓冲区
.option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576))
.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576))
// 设置 I/O处理类,主要用于网络I/O事件，记录日志，编码、解码消息
.childHandler(new CarTcpNettyChannelInitializer<SocketChannel>());
log.info("<===========netty server start success!==============>");
// 绑定端口，同步等待成功
ChannelFuture f = serverBootstrap.bind(port).sync();
// 等待服务器监听端口关闭
f.channel().closeFuture().sync();
} finally {
// 退出，释放线程池资源
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
![【web | CTF】BUUCTF [HCTF 2018]WarmUp](https://img-blog.csdnimg.cn/direct/254c6310bbd3449f9b7765a390cf7bde.png)
