Java網絡編程 — Netty入門

認識Netty

Netty簡介

Netty is?an asynchronous event-driven network application framework?for rapid development of maintainable high performance protocol servers & clients.

Netty is a NIO client server framework which enables quick and easy development of network applications such as protocol servers and clients. It greatly simplifies and streamlines network programming such as TCP and UDP socket server.

Netty是一個異步事件驅動的網絡應用框架，用于快速開發可維護的高性能協議服務器和客戶端。

Netty是一款NIO客戶端服務器框架，可以快速輕松地開發協議服務器和客戶端等網絡應用程序。它極大地簡化并簡化了TCP和UDP套接字服務器等網絡編程。

Netty架構圖如下圖所示：

Netty的特性總結如下表：

為什么選擇Netty

Netty是業界最流行的NIO框架之一，它的健壯性、功能、性能、可定制性和可擴展性在同類框架中都是首屈一指的，已經得到成百上千的商用項目驗證。通過對Netty的分析，將它的優點總結如下：

API使用簡單，開發門檻低；
功能強大，預置了多種編解碼功能，支持多種主流協議；
定制能力強，可以通過ChannelHandler對通信框架進行靈活地擴展；
性能高，通過與其他業界主流的NIO框架對比，Netty的綜合性能最優；
成熟、穩定，Netty修復了已經發現的所有JDK NIO BUG，業務開發人員不需要再為NIO的BUG而煩惱；
社區活躍，版本迭代周期短，發現的BUG可以被及時修復，同時，更多的新功能會加入；
經歷了大規模的商業應用考驗，質量得到驗證。

核心概念

Netty is a non-blocking framework. This leads to high throughput compared to blocking IO.?Understanding non-blocking IO is crucial to understanding Netty’s core components and their relationships.

Channel

Channel?is the base of Java NIO. It represents an open connection which is capable of IO operations such as reading and writing.

Channel是Java NIO的基礎。它表示一個開放的連接，能夠進行IO操作，例如讀寫。

Future

Netty中的每個IO操作都是非阻塞的，這意味著每次操作都會在通話結束后立即返回。

標準Java庫中有一個Future接口，但對Netty而言不方便 - 我們只能向Future詢問完成操作或阻止當前線程，直到完成操作。這就是為什么Netty有自己的ChannelFuture接口，我們可以將回調傳遞給ChannelFuture，其將在操作完成時被調用。

Events and Handlers

Netty使用事件驅動的應用程序范例，因此數據處理的管道是通過處理程序的一系列事件。事件和處理程序可以與入站（inbound）和出站（outbound）數據流相關聯。?Inbound events（入站事件）可以如下所示：

Channel activation and deactivation
Read operation events
Exception events
User events

Outbound events（出站事件）更簡單，通常與打開/關閉連接（?opening/closing a connection）和寫入/清空數據（writing/flushing data）有關。

Netty應用程序由幾個網絡和應用程序邏輯事件及其處理程序組成。通道事件處理程序的基礎接口是ChannelHandler及其子接口ChannelOutboundHandler``ChannelInboundHandler。

Netty提供了大量ChannelHandler實現的類。值得注意的是適配器只是空的實現，例如ChannelInboundHandlerAdapter和ChannelOutboundHandlerAdapter。

Encoders and Decoders

當我們使用網絡協議時，我們需要執行數據序列化和反序列化。為此，Netty為了能夠解碼傳入數據引入了ChannelInboundHandler的擴展解碼器，大多數解碼器的基類ByteToMessageDecoder。對于編碼輸出數據，Netty同樣提供了ChannelOutboundHandler的擴展編碼器， MessageToByteEncoder是大多數編碼器實現的基礎。

應用示例

Server Application

創建一個簡單協議服務器的項目，它接收請求，執行計算并發送響應。

Dependencies

<dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.10.Final</version>
</dependency>
復制代碼

Data Model

public class RequestData {private int intValue;private String stringValue;// standard getters and setters
}
復制代碼

public class ResponseData {private int intValue;// standard getters and setters
}
復制代碼

Request Decoder

It should be noted that Netty works with socket receive buffer, which is represented not as a queue but just as a bunch of bytes. This means that our inbound handler can be called when the full message is not received by a server.

We must make sure that we have received the full message before processing. The decoder for RequestData is shown next:

public class RequestDecoder extends ReplayingDecoder<RequestData> {private final Charset charset = Charset.forName("UTF-8");@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {RequestData data = new RequestData();data.setIntValue(in.readInt());int strLen = in.readInt();data.setStringValue(in.readCharSequence(strLen, charset).toString());out.add(data);}
}
復制代碼

ReplayingDecoder It uses an implementation of ByteBuf which throws an exception when there is not enough data in the buffer for the reading operation.

When the exception is caught the buffer is rewound to the beginning and the decoder waits for a new portion of data. Decoding stops when the out list is not empty after decode execution.

Response Encoder

public class ResponseDataEncoder extends MessageToByteEncoder<ResponseData> {@Overrideprotected void encode(ChannelHandlerContext ctx, ResponseData msg, ByteBuf out) throws Exception {out.writeInt(msg.getIntValue());}
}
復制代碼

Request Processing

public class ProcessingHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {RequestData requestData = (RequestData) msg;ResponseData responseData = new ResponseData();responseData.setIntValue(requestData.getIntValue() * 2);ChannelFuture future = ctx.writeAndFlush(responseData);future.addListener(ChannelFutureListener.CLOSE);System.out.println(requestData);}
}
復制代碼

Server Bootstrap

public class NettyServer {private int port;// constructorpublic static void main(String[] args) throws Exception {int port = args.length > 0? Integer.parseInt(args[0]);: 8080;new NettyServer(port).run();}public void run() throws Exception {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new RequestDecoder(), new ResponseDataEncoder(), new ProcessingHandler());}}).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);ChannelFuture f = b.bind(port).sync();f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}}
}
復制代碼

Client Application

The client should perform reverse encoding and decoding, so we need to have a?RequestDataEncoder?and?ResponseDataDecoder:

public class RequestDataEncoder extends MessageToByteEncoder<RequestData> {private final Charset charset = Charset.forName("UTF-8");@Overrideprotected void encode(ChannelHandlerContext ctx, RequestData msg, ByteBuf out) throws Exception {out.writeInt(msg.getIntValue());out.writeInt(msg.getStringValue().length());out.writeCharSequence(msg.getStringValue(), charset);}
}
復制代碼

public class ResponseDataDecoder extends ReplayingDecoder<ResponseData> {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {ResponseData data = new ResponseData();data.setIntValue(in.readInt());out.add(data);}
}
復制代碼

Also, we need to define a?ClientHandler?which will send the request and receive the response from server:

public class ClientHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {RequestData msg = new RequestData();msg.setIntValue(123);msg.setStringValue("all work and no play makes jack a dull boy");ChannelFuture future = ctx.writeAndFlush(msg);}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println((ResponseData)msg);ctx.close();}
}
復制代碼

Now let’s bootstrap the client:

public class NettyClient {public static void main(String[] args) throws Exception {String host = "localhost";int port = 8080;EventLoopGroup workerGroup = new NioEventLoopGroup();try {Bootstrap b = new Bootstrap();b.group(workerGroup);b.channel(NioSocketChannel.class);b.option(ChannelOption.SO_KEEPALIVE, true);b.handler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new RequestDataEncoder(), new ResponseDataDecoder(), new ClientHandler());}});ChannelFuture f = b.connect(host, port).sync();f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();}}
}
復制代碼