java架構之路-(netty專題)netty的基本使用和netty聊天室
上次回顧:
上次部落格,我們主要說了我們的IO模型,BIO同步阻塞,NIO同步非阻塞,AIO基於NIO二次封裝的非同步非阻塞,最重要的就是我們的NIO,腦海中應該有NIO的模型圖。
Netty概念:
Netty是由JBOSS提供的一個java開源框架,現為 Github上的獨立專案。Netty提供非同步的、事件驅動的網路應用程式框架和工具,用以快速開發高效能、高可靠性的網路伺服器和客戶端程式。 也就是說,Netty 是一個基於NIO的客戶、伺服器端的程式設計框架,使用Netty 可以確保你快速和簡單的開發出一個網路應用,例如實現了某種協議的客戶、服務端應用。Netty相當於簡化和流線化了網路應用的程式設計開發過程,例如:基於TCP和UDP的socket服務開發。 “快速”和“簡單”並不用產生維護性或效能上的問題。Netty 是一個吸收了多種協議(包括FTP、SMTP、HTTP等各種二進位制文字協議)的實現經驗,並經過相當精心設計的專案。最終,Netty 成功的找到了一種方式,在保證易於開發的同時還保證了其應用的效能,穩定性和伸縮性。---摘自百度百科。
基本使用:
我們來看一下程式碼吧
服務端:
package com.xiaocai.netty; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; public class NettyServer { public static void main(String[] args) throws Exception { EventLoopGroup bossGroup = new NioEventLoopGroup(1);//新建執行緒組,用於處理請求 EventLoopGroup workerGroup = new NioEventLoopGroup();//真正工作的執行緒組 try { ServerBootstrap bootstrap = new ServerBootstrap();//建立一個server,相當於NIO的server bootstrap.group(bossGroup, workerGroup) //採用鏈式程式設計,將兩個執行緒組加入到server中 .channel(NioServerSocketChannel.class) //設定一個通道,相當於NIO中的ServerSocketChannel .option(ChannelOption.SO_BACKLOG, 1024) //對服務端給予一些設定 .childHandler(new ChannelInitializer<SocketChannel>() {//給予初始化,並加入回撥函式 @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new NettyServerHandler()); } }); System.out.println("netty server start。。"); ChannelFuture cf = bootstrap.bind(9000).sync();//繫結埠,sync方法是等待非同步操作執行完畢 cf.channel().closeFuture().sync();//對通道關閉進行監聽,closeFuture是非同步操作,監聽通道關閉 // 通過sync方法同步等待通道關閉處理完畢,這裡會阻塞等待通道關閉完成 } finally { bossGroup.shutdownGracefully(); //關閉處理請求的執行緒組 workerGroup.shutdownGracefully();//關閉真正工作的執行緒組 } } }
服務端的程式碼看起來還是比較亂的,但不要擔心,基本都是固定寫法的,大致就開執行緒,設定server,設定通道,初始化,給予回撥函式,我們來看一下回調函式。
package com.xiaocai.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.util.CharsetUtil; public class NettyServerHandler extends ChannelInboundHandlerAdapter { //當客戶端連線伺服器完成就會觸發該方法 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { ByteBuf buf = Unpooled.copiedBuffer("HelloServer".getBytes(CharsetUtil.UTF_8)); ctx.writeAndFlush(buf); } //當通道有讀取事件時會觸發,即服務端傳送資料給客戶端 @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf buf = (ByteBuf) msg; System.out.println("收到客戶端的訊息:" + buf.toString(CharsetUtil.UTF_8)); System.out.println("客戶端的地址: " + ctx.channel().remoteAddress()); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { System.out.println("我收到了異常"); cause.printStackTrace(); ctx.close(); } }
這個回撥函式也基本是固定的,繼承ChannelInboundHandlerAdapter 類,然後我們寫在裡面就可以了,幾乎需要我們做的主要是這個回撥函式。
客戶端:
package com.xiaocai.netty; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; public class NettyClient { public static void main(String[] args) throws Exception { EventLoopGroup group = new NioEventLoopGroup();// 開啟工作執行緒組 try { Bootstrap bootstrap = new Bootstrap(); //建立一個和服務端相對應的server bootstrap.group(group) //設定執行緒組 .channel(NioSocketChannel.class) //使用NioSocketChannel作為客戶端的通道實現 .handler(new ChannelInitializer<SocketChannel>() {//設定回撥函式 @Override protected void initChannel(SocketChannel ch) { ch.pipeline().addLast(new NettyClientHandler()); } }); System.out.println("netty client start。。"); ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync();//啟動客戶端去連線伺服器端 //對通道關閉進行監聽 cf.channel().closeFuture().sync(); } finally { group.shutdownGracefully();//關閉執行緒組 } } }
客戶端和服務端差不多,開啟執行緒設定一個對應的server,設定通道,設定回撥函式,我們還是看一下客戶端的回撥函式做了什麼事情。
package com.xiaocai.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.util.CharsetUtil; public class NettyClientHandler extends ChannelInboundHandlerAdapter { /** * 當客戶端連線伺服器完成就會觸發該方法 * * @param ctx 設定好的通道,相當於上下文 */ @Override public void channelActive(ChannelHandlerContext ctx) { ByteBuf buf = Unpooled.copiedBuffer("HelloServer".getBytes(CharsetUtil.UTF_8)); ctx.writeAndFlush(buf); } //當通道有讀取事件時會觸發,即服務端傳送資料給客戶端 @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { ByteBuf buf = (ByteBuf) msg; System.out.println("收到服務端的訊息:" + buf.toString(CharsetUtil.UTF_8)); System.out.println("服務端的地址: " + ctx.channel().remoteAddress()); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); } }
兩側的程式碼都是差不多的,回撥函式,建立連線以後我們做什麼,接收到資料以後我們做什麼,差不多就是這個意思。這裡為什麼是給予兩個執行緒呢?簡單來說就是為了能承受更高的併發,一個執行緒專門來負責接收連線,不至於讓有的請求卡死或延遲很高的狀態,後面說底層原始碼的時候,這個還會再說的。
Netty模組元件
Bootstrap、ServerBootstrap:
Bootstrap意思是引導,一個 Netty 應用通常由一個Bootstrap 開始,主要作用是配置整個 Netty 程式,串聯各個元件,Netty 中 Bootstrap 類是客戶端程式的啟動引導類,ServerBootstrap 是服務端啟動引導類。
Future、ChannelFuture:
正如前面介紹,在 Netty 中所有的 IO 操作都是非同步的,不能立刻得知訊息是否被正確處理。 但是可以過一會等它執行完成或者直接註冊一個監聽,具體的實現就是通過 Future 和 ChannelFutures,他們可以註冊一個監聽,當操作執行成功或失敗時監聽會自動觸發註冊的監聽事 件。
Channel:
Netty 網路通訊的元件,能夠用於執行網路 I/O 操作。Channel 為使用者提供:
1)當前網路連線的通道的狀態(例如是否開啟?是否已連線?)
2)網路連線的配置引數 (例如接收緩衝區大小)
3)提供非同步的網路 I/O 操作(如建立連線,讀寫,繫結埠),非同步呼叫意味著任何 I/O 呼叫都將立即 返回,並且不保證在呼叫結束時所請求的 I/O 操作已完成。
4)呼叫立即返回一個 ChannelFuture 例項,通過註冊監聽器到 ChannelFuture 上,可以 I/O 操作成功、失敗或取消時回撥通知呼叫方。
5)支援關聯 I/O 操作與對應的處理程式。 不同協議、不同的阻塞型別的連線都有不同的 Channel 型別與之對應。
NioSocketChannel;//非同步的客戶端 TCPSocket 連線。 NioServerSocketChannel;//非同步的伺服器端 TCP Socket 連線。 NioDatagramChannel;//非同步的 UDP 連線。 NioSctpChannel;//非同步的客戶端 Sctp 連線。 NioSctpServerChannel;//非同步的 Sctp 伺服器端連線,這些通道涵蓋了UDP和TCP網路IO以及檔案IO。
Selector:
Netty 基於 Selector 物件實現 I/O 多路複用,通過 Selector 一個執行緒可以監聽多個連線的 Channel 事件。 當向一個 Selector 中註冊 Channel 後,Selector 內部的機制就可以自動不斷地查詢(Select) 這些註冊 的 Channel 是否有已就緒的 I/O 事件(例如可讀,可寫,網路連線完成等),這樣程式就可以很簡單地使用一個執行緒高效地管理多個Channel 。
NioEventLoop:
NioEventLoop 中維護了一個執行緒和任務佇列,支援非同步提交執行任務,執行緒啟動時會呼叫 NioEventLoop 的 run 方法,執行 I/O 任務和非 I/O 任務: I/O 任務,即selectionKey中ready 的事件,如 accept、connect、read、write 等,由 processSelectedKeys方法觸發。非 IO 任務,新增到 taskQueue 中的任務,如 register0、bind0 等任務,由 runAllTasks 方法觸發。
NioEventLoopGroup:
NioEventLoopGroup,主要管理 eventLoop 的生命週期,可以理解為一個執行緒池,內部維護了一組執行緒,每個執行緒(NioEventLoop)負責處理多個 Channel 上的事件,而一個 Channel 只對應於一個執行緒。
ChannelHandler:
ChannelHandler 是一個介面,處理 I/O 事件或攔截 I/O 操作,並將其轉發到其 ChannelPipeline(業務處理鏈)中的下一個處理程式。 ChannelInboundHandler 用於處理入站 I/O 事件,ChannelOutboundHandler用於處理出站 I/O事件。
ChannelHandlerContext:
儲存 Channel 相關的所有上下文資訊,同時關聯一個 ChannelHandler 物件。
ChannelPipline:
儲存 ChannelHandler 的List,用於處理或攔截 Channel 的入站事件和出站操作。內部是一個雙向連結串列,記住tail和head,後面說編解碼會詳細去說。
netty聊天室:
網際網路行業:在分散式系統中,各個節點之間需要遠端服務呼叫,高效能的 RPC 框架必不可少, Netty 作為非同步高效能的通訊框架,往往作為基礎通訊元件被這些 RPC 框架使用。典型的應用有:阿里分散式服務框架 Dubbo 的 RPC 框架使用 Dubbo 協議進行節點間通訊,Dubbo 協議預設使用 Netty 作為基礎通訊元件,用於實現。各程序節點之間的內部通訊。Rocketmq底層也是用的Netty作為基礎通訊元件。 遊戲行業:無論是手遊服務端還是大型的網路遊戲,Java 語言得到了越來越廣泛的應用。Netty 作為高效能的基礎通訊元件,它本身提供了 TCP/UDP 和 HTTP 協議棧。我們來實現一個簡單的聊天室。一步步的我們拿著剛才的程式碼改造一下。
服務端:
package com.xiaocai.chat; import com.xiaocai.netty.NettyServerHandler; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; public class ChatRoomServer { public static void main(String[] args) throws Exception { EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new StringEncoder()); ch.pipeline().addLast(new StringDecoder()); ch.pipeline().addLast(new ChatRoomServerHandler()); } }); System.out.println("聊天室開啟了!!!"); ChannelFuture cf = bootstrap.bind(9000).sync(); cf.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }
程式還是和原來的程式差不多的,這裡都是一些固定的寫法,只有兩行程式碼我們剛才沒說,加入了兩個攔截器,一個是編碼一個是解碼,下次部落格會重點說這個,我們來看一下我們的回撥函式。
package com.xiaocai.chat; import io.netty.channel.*; import io.netty.channel.group.ChannelGroup; import io.netty.channel.group.DefaultChannelGroup; import io.netty.util.concurrent.GlobalEventExecutor; import java.text.SimpleDateFormat; import java.util.Date; public class ChatRoomServerHandler extends SimpleChannelInboundHandler<String> { SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); //當客戶端連線伺服器完成就會觸發該方法 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); String nowTime = sdf.format(new Date()); String userName = channel.id() + ""; String msg = nowTime + " " + userName + " 上線了!"; channelGroup.writeAndFlush(msg); channelGroup.add(channel); System.out.println("[服務端]: " + msg); } @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { //獲取到當前 channel Channel channel = ctx.channel(); //這時我們遍歷 channelGroup, 根據不同的情況, 回送不同的訊息 channelGroup.forEach(ch -> { if (channel != ch) { //不是當前的 channel,轉發訊息 ch.writeAndFlush(channel.id() + " 傳送了訊息:" + msg); } else {//回顯自己傳送的訊息給自己 ch.writeAndFlush("[ 自己 ]傳送了訊息:" + msg); } }); } //表示 channel 處於不活動狀態, 提示離線了 @Override public void channelInactive(ChannelHandlerContext ctx) { Channel channel = ctx.channel(); //將客戶離開資訊推送給當前線上的客戶 channelGroup.writeAndFlush("[ 客戶端 ]" + channel.id() + " 下線了"); System.out.println(channel.id() + " 下線了"); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { //關閉通道 ctx.close(); } }
這次我們繼承的是SimpleChannelInboundHandler這個類,他是對於ChannelInboundHandlerAdapter的再次封裝,然後我們的是通過通道組來發送訊息的,先來看channelActive方法,就是當有客戶端連線的時候,我們應該自己有所感知,並且將某某上線的訊息廣播出去。channelRead0方法就是當有資料傳輸過來時的一些處理,我們用channelGroup來判斷是否是自己傳送的訊息,是自己傳送的訊息我們給予自己說了什麼什麼,別的訊息就是某某某說了什麼什麼。
客戶端:
package com.xiaocai.chat; import io.netty.bootstrap.Bootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; import java.util.Scanner; public class ChatRoomClient { public static void main(String[] args) throws Exception { EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap(); bootstrap.group(group) //設定執行緒組 .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() {//設定回撥函式 @Override protected void initChannel(SocketChannel ch) { ch.pipeline().addLast(new StringEncoder()); ch.pipeline().addLast(new StringDecoder()); ch.pipeline().addLast(new ChatRoomClientHandler()); } }); ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync(); //得到 channel Channel channel = cf.channel(); System.out.println("我" + channel.id() + "進入了聊天室"); //客戶端需要輸入資訊, 建立一個掃描器 Scanner scanner = new Scanner(System.in); while (scanner.hasNextLine()) { String msg = scanner.nextLine(); //通過 channel 傳送到伺服器端 channel.writeAndFlush(msg); } cf.channel().closeFuture().sync(); } finally { group.shutdownGracefully(); } } }
客戶端的程式碼和原來的也是差不多了,還是多兩個攔截器,進行編碼解碼,然後下面就是我們的掃描器,來掃描控制檯輸入的,一會我們要用控制檯來完成這個聊天室,所以我們要掃描控制檯的輸入,我們來看一下我們的攔截器。
package com.xiaocai.chat; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; public class ChatRoomClientHandler extends SimpleChannelInboundHandler<String> { @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { System.out.println(msg); } }
就是當有訊息發過來時,我們打印出來就可以了,我們來看看我們的效果。
大概就是這樣的,就是我們的客戶端傳送訊息,由服務端接收,然後由服務端再次轉發出去。
總結:
今天我們主要說了Netty的基本使用,需要記住那幾行固定程式碼都是什麼意思,為什麼要開啟兩個執行緒來共同處理,然後就是我們的一個小例子,聊天室程式,下次我們來說一下我們的編碼解碼,還有粘包拆包。
最進弄了一個公眾號,小菜技術,歡迎大家的加入