Netty學習之路(四)-Netty入門實戰
阿新 • • 發佈:2018-12-05
前面學習了用Java原生NIO的程式設計實踐,過程還是挺複雜的,需要熟練掌握Selector,ServerSocketChannel握,SocketChannel,ByteBuffer等。所以在絕大多數業務場景中我們可以使用Netty來進行NIO程式設計。先總結一下Netty的優點:
- API使用簡單,開發門檻低
- 功能強大,預製了多種編解碼功能,支援多種主流協議
- 定製能力強,可以通過ChannelHandler對通訊框架進行靈活的擴充套件
- 效能高,成熟,穩定,社群活躍,版本迭代週期短
- 經歷了大規模的商業應用考驗,質量得到驗證等
至於安裝就不多說了,只要下載他的JAR包然後在普通java專案中匯入就可以了。
程式設計實戰
可以對比一下之前的原生NIO程式碼,是簡潔了許多。
Netty服務端
package com.ph.Netty; import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; /** * Create by PH on 2018/11/3 */ public class NettyServer { public static void main(String[] args) throws Exception { int port = 8080; if(args !=null && args.length>0) { try { port = Integer.valueOf(args[0]); }catch (NumberFormatException e) { //採用預設值 } } new NettyServer().bind(port); } public void bind(int port) throws Exception{ //NioEventLoopGroup是一個執行緒組,包含了一組NIO執行緒,專門用於網路事件的處理, //實際上他們就是Reactor執行緒組 //bossGroup僅接收客戶端連線,不做複雜的邏輯處理,為了儘可能減少資源的佔用,取值越小越好 EventLoopGroup bossGroup = new NioEventLoopGroup(1); //用於進行SocketChannel的網路讀寫 EventLoopGroup workerGroup = new NioEventLoopGroup(); try { //是Netty用於啟動NIO服務端的輔助啟動類,目的是降低服務端的開發複雜度 ServerBootstrap b = new ServerBootstrap(); //配置NIO服務端 b.group(bossGroup, workerGroup) //指定使用NioServerSocketChannel產生一個Channel用來接收連線,他的功能對應於JDK // NIO類庫中的ServerSocketChannel類。 .channel(NioServerSocketChannel.class) //BACKLOG用於構造服務端套接字ServerSocket物件,標識當伺服器請求處理執行緒全滿時, // 用於臨時存放已完成三次握手的請求的佇列的最大長度。如果未設定或所設定的值小於1, // Java將使用預設值50。 .option(ChannelOption.SO_BACKLOG, 1024) //繫結I/O事件處理類,作用類似於Reactor模式中的Handler類,主要用於處理網路I/O事件 .childHandler(new ChildChannelHandler()); //繫結埠,同步等待繫結操作完成,完成後返回一個ChannelFuture,用於非同步操作的通知回撥 ChannelFuture f = b.bind(port).sync(); //等待服務端監聽埠關閉之後才退出main函式 f.channel().closeFuture().sync(); } finally { //退出,釋放執行緒池資源 bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } private class ChildChannelHandler extends ChannelInitializer<SocketChannel> { protected void initChannel(SocketChannel arg0) throws Exception { arg0.pipeline().addLast(new ServerHandler()); } } } /** * ChannelInboundHandlerAdapter實現自ChannelInboundHandler * ChannelInboundHandler提供了不同的事件處理方法可通過重寫來自定義處理方式 */ class ServerHandler extends ChannelInboundHandlerAdapter { /** * 接受客戶端傳送的訊息 * @param ctx * @param msg * @throws Exception */ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //類似JDK中的ByteBuffer物件,不過它提供了更加強大和靈活的功能 ByteBuf buf = (ByteBuf) msg; //通過readableBytes()方法獲取緩衝區可讀的位元組數 byte[] req = new byte[buf.readableBytes()]; //將緩衝區中的位元組陣列複製到新建的byte陣列中 buf.readBytes(req); String body = new String(req, "UTF-8"); System.out.println("Server receive: " + body); //獲得ByteBuf型別的資料 ByteBuf resp = Unpooled.copiedBuffer("Server message".getBytes()); //向客戶端傳送訊息,不直接將訊息寫入SocketChannel中,只是把待發送的訊息放到傳送快取陣列中, //再通過呼叫flush方法將緩衝區中的訊息全部寫到SocketChannel中 ctx.write(resp); } public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { //將訊息傳送佇列中的訊息寫入到SocketChannel中傳送給對方 ctx.flush(); } public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { //當發生異常時釋放資源 ctx.close(); } }
Netty客戶端
package com.ph.Netty; import io.netty.bootstrap.Bootstrap; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; /** * Create by PH on 2018/11/3 */ public class NettyClient { public static void main(String[] args) throws Exception { int port = 8080; if (args != null && args.length > 0) { try { port = Integer.valueOf(args[0]); } catch (NumberFormatException e) { //採用預設值 } } new NettyClient().connect(port, "127.0.0.1"); } public void connect(int port, String host) throws Exception{ //配置客戶端NIO執行緒組 EventLoopGroup group = new NioEventLoopGroup(); try{ Bootstrap b = new Bootstrap(); b.group(group).channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) .handler(new ChannelInitializer<SocketChannel>() { public void initChannel(SocketChannel ch) throws Exception{ ch.pipeline().addLast(new ClientHandler()); } }); //發起非同步連線操作 ChannelFuture f = b.connect(host, port).sync(); //等待客戶端鏈路關閉 f.channel().closeFuture().sync(); }finally { group.shutdownGracefully(); } } } class ClientHandler extends ChannelInboundHandlerAdapter { private final ByteBuf msg; public ClientHandler() { byte[] req = "Client message".getBytes(); msg = Unpooled.buffer(req.length); msg.writeBytes(req); } /** * 當客戶端和服務端TCP鏈路建立成功之後,Netty的NIO執行緒會呼叫此方法 * @param ctx */ public void channelActive(ChannelHandlerContext ctx) { //傳送訊息到服務端 ctx.writeAndFlush(msg); } public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception{ ByteBuf buf = (ByteBuf) msg; byte[] req = new byte[buf.readableBytes()]; buf.readBytes(req); String body = new String(req, "utf-8"); System.out.println("Client receive :" + body); } public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { ctx.close(); } }