Netty 是一个事件驱动的异步网络框架。
重要的四个学习资料：
- github：https://github.com/netty/netty
- 《Netty 权威指南》
- 《Java 读源码之 Netty 深入剖析》
本文会提供一个 Netty 最简使用示例，之后的源码分析都会基于该示例及其扩展进行。关于 Netty 的最佳实践见 SOFABolt 源码分析，本系列也会介绍 Netty 在 SOFABolt 中的使用姿势。
Netty 坐标：
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.33.Final</version>
</dependency>
一、服务端
1.1 服务端
public final class EchoServer {
public static void main(String[] args) throws Exception {
// 1. 配置 bossGroup 和 workerGroup
final EventLoopGroup bossGroup = new NioEventLoopGroup(1);
final EventLoopGroup workerGroup = new NioEventLoopGroup();
// 2. 创建业务逻辑处理器
final EchoServerHandler serverHandler = new EchoServerHandler();
// 3. 创建并配置服务端启动辅助类 ServerBootstrap
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 100)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(serverHandler);
}
});
// 4. 阻塞绑定端口
ChannelFuture f = b.bind(8081).sync();
// 5. 为服务端关闭的 ChannelFuture 添加监听器，用于实现优雅关闭
f.channel().closeFuture().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
});
}
}
建议：
- 如果 bossGroup 只用于绑定一个端口，那么 groupSize 设为1，因为只会使用到 bossGroup 中的一个 NioEventLoop，如果 groupSize>1，则其他多余的 NioEventLoop 只会白占内存（后续会进行源码分析）
- 建议使用异步的方式进行优雅关闭，不建议
f.channel().closeFuture().sync();同步关闭，因为在实际使用中，主线程还需要继续往后执行，不能阻塞。
1.2 服务端业务逻辑处理器
@Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
// 1. 将接收到的信息强转为 ByteBuf
ByteBuf buf = (ByteBuf)msg;
// 2. 创建 byte[]
byte[] req = new byte[buf.readableBytes()];
// 3. 将 ByteBuf 中的字节写入 byte[]
buf.readBytes(req);
// 4. 将 byte[] 转化为 string，并输出
System.out.println("server received: " + new String(req));
// 5. 创建响应的 ByteBuf，回写接收到的信息
ByteBuf resp = Unpooled.wrappedBuffer(req);
// 6. 将响应消息写入发送缓冲数组
ctx.write(resp);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
// 将发送缓冲数组中的消息全部冲刷到 SocketChannel 中
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
// Close the connection when an exception is raised.
cause.printStackTrace();
ctx.close();
}
}
二、客户端
2.1 客户端
public final class EchoClient {
// 1. 配置 EventLoopGroup
final static EventLoopGroup group = new NioEventLoopGroup();
public static void main(String[] args) throws Exception {
// 2. 创建并配置客户端启动辅助类 Bootstrap
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new EchoClientHandler());
}
});
// 3. 阻塞连接服务端
ChannelFuture f = b.connect("127.0.0.1", 8081).sync();
f.channel().closeFuture().addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
group.shutdownGracefully();
}
});
}
}
最佳实践：通常一个服务可能会调用多个其他服务，针对每一个被调服务，会创建一个 NettyClient 与被调服务的 NettyServer 进行通信，为了防止 NioEventLoop 太多，多个 NettyClient 可以共享一个 NioEventLoopGroup。
2.2 客户端业务逻辑处理器
public class EchoClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) {
ByteBuf buf = Unpooled.wrappedBuffer("hello,netty".getBytes());
ctx.writeAndFlush(buf);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf buf = (ByteBuf) msg;
byte[] resp = new byte[buf.readableBytes()];
buf.readBytes(resp);
System.out.println("client received: " + new String(resp));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
// Close the connection when an exception is raised.
cause.printStackTrace();
ctx.close();
}
}
整个交互流程
- 服务端启动并绑定端口，之后启动一条 NIO 线程监听客户端连接；
- 客户端进行阻塞连接，连接完成时，创建 ByteBuf，发送数据给服务端；
- 服务端接收到消息，将数据打印到 console 之后，原封数据返回给客户端；
- 客户端读取数据并打印到 console，完成一次 ping-pong
