概述
Netty是由JBOSS提供的一个java开源框架，现为 Github上的独立项目。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。
它是一个NIO客户端服务器框架，可以快速轻松地开发网络应用程序，极大地简化了网络编程，例如TCP和UDP套接字服务器开发。
整合
在pom.xml中添加依赖：
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.42.Final</version>
</dependency>
️以下使用时间服务器作为例子。
服务端实现
创建NettyServer类添加@Component注解交由容器管理
@Component
public class NettyServer {
private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);
// 服务端NIO线程组
private final EventLoopGroup bossGroup = new NioEventLoopGroup();
private final EventLoopGroup workGroup = new NioEventLoopGroup();
public ChannelFuture start(String host, int port) {
ChannelFuture channelFuture = null;
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
// 自定义服务处理
socketChannel.pipeline().addLast(new ServerHandler());
}
});
// 绑定端口并同步等待
channelFuture = bootstrap.bind(host, port).sync();
logger.info("======Start Up Success!=========");
} catch (Exception e) {
e.printStackTrace();
}
return channelFuture;
}
public void close() {
workGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
logger.info("======Shutdown Netty Server Success!=========");
}
}
创建服务处理类ServerHandler继承ChannelInboundHandlerAdapter类并重写channelRead、channelReadComplete、exceptionCaught方法。
public class ServerHandler extends ChannelInboundHandlerAdapter {
/**
* 客户端数据到来时触发
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
System.out.println("client request: " + buf.toString(CharsetUtil.UTF_8));
SimpleDateFormat sf = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
String callback = sf.format(new Date());
ctx.write(Unpooled.copiedBuffer(callback.getBytes()));
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
// 将发送缓冲区的消息全部写到SocketChannel中
ctx.flush();
}
/**
* 发生异常时触发
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
// 释放与ChannelHandlerContext相关联的资源
ctx.close();
}
}
在SpringBoot启动类上实现CommandLineRunner接口。
注入容器中的NettyServer对象，在run方法中添加需要监听的地址以及端口开启服务。
获取当前线程钩子，使jvm关闭前服务同时关闭释放相关资源。
@SpringBootApplication
public class NioApplication implements CommandLineRunner {
public static void main(String[] args) {
SpringApplication.run(NioApplication.class, args);
}
@Resource
private NettyServer nettyServer;
@Override
public void run(String... args) throws Exception {
// 开启服务
ChannelFuture future = nettyServer.start("localhost", 7070);
// 在JVM销毁前关闭服务
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
nettyServer.close();
}
});
future.channel().closeFuture().sync();
}
}
客户端实现
新建一个普通的项目并添加依赖netty-all-4.1.43.Final.jar（下载并解压后只需添加all-in-one文件夹内的netty-all-4.1.43.Final.jar）
创建NettyClient类作为客户端
public class NettyClient {
private static final EventLoopGroup group = new NioEventLoopGroup();
public static void main(String[] args) throws Exception {
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
// 自定义处理程序
socketChannel.pipeline().addLast(new ClientHandler());
}
});
// 绑定端口并同步等待
ChannelFuture channelFuture = bootstrap.connect("localhost", 7070).sync();
channelFuture.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}
创建处理程序类ClientHandler继承ChannelInboundHandlerAdapter类重写channelActive、channelRead、exceptionCaught方法。
public class ClientHandler extends ChannelInboundHandlerAdapter {
/**
* 连接到服务器时触发
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(Unpooled.copiedBuffer("current time", CharsetUtil.UTF_8));
}
/**
* 消息到来时触发
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
System.out.println("Current Time: " + buf.toString(CharsetUtil.UTF_8));
}
/**
* 发生异常时触发
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
测试
以上一个简单的服务端以及客户端就配置完成了，分别运行服务端程序以及客户端程序进行测试。
Server-1
Client-1
粘包处理
经过测试一个简单的时间服务器配置成功，现在修改客户端，使客户端能够同时发送大量数据，并进行测试。
修改客户端处理程序ClientHandler类中的channelActive方法，连续发送100条信息。
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
for (int i = 0; i < 100; i++) {
ctx.write(Unpooled.copiedBuffer("current time", CharsetUtil.UTF_8));
}
ctx.flush();
}
服务端接收如下所示：
Server-2
客户端接收如下所示：
Client-2
由图可知服务端只接收了2次，并且数据被连接在了一起，可知发生了粘包的情况。
下面我们使用netty提供的解码器来解决粘包问题。
修改发送的数据格式，在数据之间添加换行符 \n 进行间隔：
//--------ClientHandler---------
ctx.write(Unpooled.copiedBuffer("current time\n", CharsetUtil.UTF_8));
//--------ServerHandler---------
String callback = sf.format(new Date()) + "\n";
在之前配置的NettyServer类的服务处理程序之前添加解码器LineBasedFrameDecoder：
bootstrap.group(bossGroup, workGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
// LineBasedFrameDecoder能够将接收到的数据在行尾进行拆分。
// 设置解码帧的最大长度，如果帧的长度超过此值抛出异常
socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
// 服务处理
socketChannel.pipeline().addLast(new ServerHandler());
}
});
同理客户端类NettyClient也同时添加解码器LineBasedFrameDecoder：
bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
socketChannel.pipeline().addLast(new ClientHandler());
}
});
再次进行测试
Server-3
Client-3
