Netty是Java領域有名的開源網絡庫，特點是高性能和高擴展性，因此很多流行的框架都是基於它來構建的，比如我們熟知的Dubbo、Rocketmq、Hadoop等，針對高性能RPC，一般都是基於Netty來構建，比如sock-bolt。總之一句話，Java小伙伴們需要且有必要學會使用Netty並理解其實現原理。
netty旨在為可維護的高性能、高可擴展性協議服務器和客戶端的快速開發提供異步事件驅動的網絡應用程序框架和工具。換句話說，Netty是一個NIO客戶端服務器框架，可以快速輕松地開發協議服務器和客戶端等網絡應用程序。它極大地簡化並簡化了TCP和UDP套接字服務器開發等網絡編程。
學習netty原理細節，看netty源碼是必不可少的，那首先來看下如何編譯源碼：
- 從github下載netty 4.x源碼
- 如果缺少XxxObjectHashMap類，這些類是在編譯時自動生成的，可以執行mvn clean install或者cd common && mvn clean install命令即可。
- 打開idea，開啟源碼閱讀之旅 😃
除了看源碼，可以結合一些書籍來看，學習效果更好。關於Netty的書籍，筆者這里推薦一本 李林鋒 寫的《Netty權威指南》，這本書對於Netty的基礎概念和NIO部分講解的還是不錯的，不過有點地方感覺有點貼代碼湊字數嫌疑，整體來說還算不錯。
什么是Netty
Netty是一個事件驅動的高性能Java網絡庫，是一個隱藏了背后復雜性而提供一個易於使用的API的客戶端/服務端框架。Netty以其高性能和可擴展性，使開發者專注於真正感興趣的地方。它的一個主要目標就是促進“關注點分離”：使業務邏輯從網絡基礎設施應用程序中分離。
不僅僅是Netty框架，其他框架的設計目的也大都是為了使業務程序和底層技術解耦，使程序員更加專注於業務邏輯實現，提高開發質量和效率。Netty為什么性能如此之高，主要是其內部的Reactor模型機制。
Netty核心組件
- Bootstrap和ServerBootstrap：Netty應用程序通過設置bootstrap引導類來完成，該類提供了一個用於應用程序網絡層配置的容器。Bootstrap服務端的是ServerBootstrap，客戶端的是Bootstrap。
- Channel：Netty 中的接口 Channel 定義了與 socket 豐富交互的操作集：bind, close, config, connect, isActive, isOpen, isWritable, read, write 等等。
- ChannelHandler：ChannelHandler 支持很多協議，並且提供用於數據處理的容器，ChannelHandler由特定事件觸發， 常用的一個接口是ChannelInboundHandler，該類型處理入站讀數據（socket讀事件）。
- ChannelPipeline：ChannelPipeline 提供了一個容器給 ChannelHandler 鏈並提供了一個API 用於管理沿着鏈入站和出站事件的流動。每個 Channel 都有自己的ChannelPipeline，當 Channel 創建時自動創建的。 下圖說明了ChannelHandler和ChannelPipeline二者的關系：
- EventLoop：EventLoop 用於處理 Channel 的 I/O 操作。一個單一的 EventLoop通常會處理多個 Channel 事件。一個 EventLoopGroup 可以含有多於一個的 EventLoop 和 提供了一種迭代用於檢索清單中的下一個。
- ChannelFuture：Netty 所有的 I/O 操作都是異步。因為一個操作可能無法立即返回，我們需要有一種方法在以后獲取它的結果。出於這個目的，Netty 提供了接口 ChannelFuture,它的 addListener 方法
Netty 是一個非阻塞、事件驅動的網絡框架。Netty 實際上是使用 Threads（ 多線程） 處理 I/O事件的，對於熟悉多線程編程的讀者可能會需要關注同步代碼。這樣的方式不好，因為同步會影響程序的性能，Netty 的設計保證程序處理事件不會有同步。因為某個Channel事件是被添加到一個EventLoop中的，以后該Channel事件都是由該EventLoop來處理的，而EventLoop是一個線程來處理的，也就是說Netty不需要同步IO操作，EventLoop與EventLoopGroup的關系可以理解為線程與線程池的關系一樣。
Buffer（緩沖）
ByteBuf是字節數據的容器，所有的網絡通信都是基於底層的字節流傳輸，ByteBuf 是一個很好的經過優化的數據容器，我們可以將字節數據有效的添加到 ByteBuf 中或從 ByteBuf 中獲取數據。為了便於操作，ByteBuf 提供了兩個索引：一個用於讀，一個用於寫。我們可以按順序讀取數據，也可以通過調整讀取數據的索引或者直接將讀取位置索引作為參數傳遞給get方法來重復讀取數據。
ByteBuf使用模式
堆緩沖區ByteBuf將數據存儲在 JVM 的堆空間，這是通過將數據存儲在數組的實現。堆緩沖區可以快速分配，當不使用時也可以快速釋放。它還提供了直接訪問數組的方法，通過 ByteBuf.array() 來獲取 byte[]數據。
堆緩沖區ByteBuf使用示例：
ByteBuf heapBuf = ...;
if (heapBuf.hasArray()) {
byte[] array = heapBuf.array();
int offset = heapBuf.arrayOffset() + heapBuf.readerIndex();
int length = heapBuf.readableBytes();
handleArray(array, offset, length);
}
直接緩沖區ByteBuf，在 JDK1.4 中被引入 NIO 的ByteBuffer 類允許 JVM 通過本地方法調用分配內存，其目的是通過免去中間交換的內存拷貝, 提升IO處理速度; 直接緩沖區的內容可以駐留在垃圾回收掃描的堆區以外。DirectBuffer 在 -XX:MaxDirectMemorySize=xxM大小限制下, 使用 Heap 之外的內存, GC對此”無能為力”，也就意味着規避了在高負載下頻繁的GC過程對應用線程的中斷影響。
Netty示例代碼
了解了Netty基礎概念之后，一起看下Netty的使用示例，下面以TCP server、TCP client、http server為例，由於示例代碼不難，所以不再贅述，直接上代碼。
TCP Server
public static void main(String[] args) {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap boot = new ServerBootstrap();
boot.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.localAddress(8080)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new EchoHandler());
}
});
// start
ChannelFuture future = boot.bind().sync();
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
// shutdown
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public class EchoHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf in = (ByteBuf) msg;
System.out.println(in.toString(CharsetUtil.UTF_8));
ctx.write(msg);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
TCP client
public static void main(String[] args) {
EventLoopGroup group = new NioEventLoopGroup();
try {
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 LoggingHandler(LogLevel.INFO));
p.addLast(new EchoClientHandler());
}
});
// Start the client.
ChannelFuture f = b.connect("localhost", 8081).sync();
f.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
group.shutdownGracefully();
}
}
public class EchoClientHandler extends ChannelInboundHandlerAdapter {
private final ByteBuf message;
public EchoClientHandler() {
message = Unpooled.buffer(256);
message.writeBytes("hello netty".getBytes(CharsetUtil.UTF_8));
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
ctx.writeAndFlush(message);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
System.out.println(((ByteBuf) msg).toString(CharsetUtil.UTF_8));
ctx.write(msg);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
// Close the connection when an exception is raised.
cause.printStackTrace();
ctx.close();
}
}
netty client端在什么時候將channel注冊到selector上的呢？是在創建channel之后，就注冊到selector的，相關代碼在initAndRegister方法中：
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
// 創建(netty自定義)Channel實例，並初始化
// channel為 NioServerSocketChannel 實例,NioServerSocketChannel的父類AbstractNioChannel保存有nio的ServerSocketChannel
channel = channelFactory.newChannel();
init(channel);
} catch (Throwable t) {
if (channel != null) {
// channel can be null if newChannel crashed (eg SocketException("too many open files"))
channel.unsafe().closeForcibly();
// as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
}
// as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
}
// 向Selector注冊channel
ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
// If we are here and the promise is not failed, it's one of the following cases:
// 1) If we attempted registration from the event loop, the registration has been completed at this point.
// i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
// 2) If we attempted registration from the other thread, the registration request has been successfully
// added to the event loop's task queue for later execution.
// i.e. It's safe to attempt bind() or connect() now:
// because bind() or connect() will be executed *after* the scheduled registration task is executed
// because register(), bind(), and connect() are all bound to the same thread.
return regFuture;
}
initAndRegister之后會執行connect動作，注意，真正的channel.connect動作是由NioEventLoop線程來完成的，當連接三次握手完成之后，會觸發該channel的ACCEPT事件，也就是NIOEventLoop中處理事件的流程。
Http server
public static void main(String[] args) {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap boot = new ServerBootstrap();
boot.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.localAddress(8080)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast("decoder", new HttpRequestDecoder())
.addLast("encoder", new HttpResponseEncoder())
.addLast("aggregator", new HttpObjectAggregator(512 * 1024))
.addLast("handler", new HttpHandler());
}
});
// start
ChannelFuture future = boot.bind().sync();
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
// shutdown
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public class HttpHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
HttpResponseStatus.OK,
Unpooled.wrappedBuffer("hello netty".getBytes()));
HttpHeaders heads = response.headers();
heads.add(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN + "; charset=UTF-8");
heads.add(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes()); // 3
heads.add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
ctx.writeAndFlush(response);
}
}
推薦閱讀
歡迎小伙伴關注【TopCoder】閱讀更多精彩好文。
