一、為何要使用netty開(kāi)發(fā)

由于之前已經(jīng)用Java中的socket寫(xiě)過(guò)一版簡(jiǎn)單的聊天室，這里就不再對(duì)聊天室的具體架構(gòu)進(jìn)行細(xì)致的介紹了，主要關(guān)注于使用netty框架重構(gòu)后帶來(lái)的改變。對(duì)聊天室不了解的同學(xué)可以先看下我的博客（《JAVA簡(jiǎn)單聊天室的實(shí)現(xiàn)》）

本篇博客所使用的netty版本為4.1.36，完整工程已上傳到Github（https://github.com/Alexlingl/Chatroom），其中l(wèi)ib文件夾下有相應(yīng)的netty jar包和source包，自行導(dǎo)入即可。

1、為何要重構(gòu)

之前的聊天室是基于Java原生socket實(shí)現(xiàn)的，socket的處理機(jī)制屬于BIO模型，也就是阻塞IO模型。對(duì)于每一個(gè)客戶端的連接我們都需要啟動(dòng)一個(gè)線程來(lái)處理，并且該線程會(huì)一直阻塞在讀取用戶數(shù)據(jù)上面。如此一來(lái)，一旦有大量的客戶端并發(fā)連接我們的服務(wù)器，服務(wù)器將難以承受。之前用JMeter測(cè)試過(guò)，單純使用Java原生socket開(kāi)發(fā)的服務(wù)器所能支持的最大并發(fā)在2300左右。雖然采用線程池的策略可以在一定程度上提升最大并發(fā)數(shù)，但也無(wú)法超過(guò)1W。因此我們需要對(duì)其進(jìn)行重構(gòu)，使其能夠具有更高的性能。

2、為何使用netty框架

使用netty框架主要還是為了提升代碼的開(kāi)發(fā)速度，并且減少代碼維護(hù)成本。使用netty框架開(kāi)發(fā)的程序在復(fù)雜度上比使用Java原生NIO類庫(kù)開(kāi)發(fā)的要小很多。具體可以看下我之前關(guān)于解決C10k問(wèn)題的系列文章，里面有具體的代碼。

3、為何netty框架只實(shí)現(xiàn)了NIO而沒(méi)有AIO

前面在解決C10問(wèn)題時(shí)，探究過(guò)NIO和AIO的區(qū)別，并且使用Java所提供的類庫(kù)實(shí)現(xiàn)了兩個(gè)小程序，理論上來(lái)說(shuō)AIO性能明顯要比NIO高，那為什么netty使用了NIO而不是AIO呢？

官方說(shuō)法如下：

We obviously did not consider Windows as a serious platform so far, and that’s why we were neglecting NIO.2 AIO API which was implemented using IOCP on Windows. (On Linux, it wasn’t any faster because it was using the same OS facility - epoll.)

大意就是，windows上面有IOCP來(lái)支持AIO的實(shí)現(xiàn)，因此AIO的性能會(huì)比NIO好。而Linux上面不管是NIO還是AIO，底層都是用epoll實(shí)現(xiàn)的，性能差距不大。然而當(dāng)下絕大部分的服務(wù)器還是建立在Linux上，因此沒(méi)必要使用AIO（使用AIO反而會(huì)增加代碼的復(fù)雜度，增大維護(hù)成本）。

二、基于netty NIO的處理模型

1、服務(wù)器的類關(guān)系

（1）、SubreqServer：創(chuàng)建兩個(gè)NIO線程組，一個(gè)用來(lái)監(jiān)聽(tīng)處理客戶端的連接請(qǐng)求，一個(gè)用來(lái)監(jiān)聽(tīng)客戶端的消息。同時(shí)實(shí)例化一個(gè)ServerBootStrap啟動(dòng)類的對(duì)象來(lái)啟動(dòng)兩個(gè)NIO線程組，并且配置必要的參數(shù)。（2）、ChannelInitializer：初始化SocketChannel管道的各項(xiàng)參數(shù)，主要有指定解碼器和編碼器，并指明管道的處理類（3）、SubreqServerHandler：SocketChannel管道的處理類，負(fù)責(zé)處理來(lái)自客戶端的消息

2、客戶端的類關(guān)系

（1）、SubreqClient：創(chuàng)建一個(gè)NIO線程組，用來(lái)監(jiān)聽(tīng)客戶端的消息。同時(shí)實(shí)例化一個(gè)ServerBootStrap啟動(dòng)類的對(duì)象來(lái)啟動(dòng)這個(gè)NIO線程組，并且配置必要的參數(shù)。最后讓主程序阻塞在監(jiān)聽(tīng)客戶端的鍵盤輸入。（2）、ChannelInitializer：初始化SocketChannel管道的各項(xiàng)參數(shù)，主要有指定解碼器和編碼器，并指明管道的處理類（3）、SubreqClientHandler：SocketChannel管道的處理類，負(fù)責(zé)處理來(lái)自服務(wù)器的消息

三、所涉及類庫(kù)的源碼解讀

1、ChannelHandlerAdapter（用來(lái)對(duì)channel的注冊(cè)和注銷做出反應(yīng)的類）：

（1）、功能

用來(lái)實(shí)現(xiàn)當(dāng)用戶上線或下線時(shí)，通知其他在線的用戶。

（2）、類定義

它是ChannelHandler的框架實(shí)現(xiàn)

（3）、HandlerAdded()和HandlerRemoved()方法

當(dāng)有一個(gè)channel被注冊(cè)后將會(huì)調(diào)用HandlerAdded()，而當(dāng)有一個(gè)channel被注銷后將調(diào)用HandlerRemoved()方法。并且根據(jù)注釋我們知道，這兩個(gè)方法默認(rèn)不做任何處理，它希望由繼承的子類自己去寫(xiě)相應(yīng)的處理實(shí)現(xiàn)。

2、SimpleChannelInboundHandler（用來(lái)對(duì)接收的消息做出反應(yīng)）

（1）、功能

用來(lái)實(shí)現(xiàn)當(dāng)接收到消息時(shí)做出相應(yīng)反應(yīng)，如果是服務(wù)端，那么將當(dāng)前消息轉(zhuǎn)發(fā)給其他在線的客戶端；如果是客戶端，就將消息簡(jiǎn)單地打印出來(lái)。

（2）、類定義

這個(gè)類是一個(gè)泛型類，它只能用于處理一種具體類型的消息。注釋中給出一種使用方法，StringHandler繼承了SimpleChannelInboundHandler，并且指定泛型變量為String，因此這個(gè)繼承類只能用于String類型消息的處理。

（2）、channelRead(ChannelHandlerContext ctx, Object msg)

這個(gè)方法主要用來(lái)處理類型為Object的消息，也就是所有消息。

首先當(dāng)接收到msg時(shí)，先使用accpetInboundMessage()方法來(lái)判斷該消息是否可以處理。我們來(lái)看下該方法的實(shí)現(xiàn)。

如果接收到的這個(gè)消息應(yīng)該被處理就返回true，如果它應(yīng)當(dāng)被傳到ChannelPipeLine的下一個(gè)ChannelInboundHandler就返回false。

我們繼續(xù)來(lái)看下ChannelRead()方法。如果接受到的消息可以處理時(shí)。它將對(duì)消息進(jìn)行強(qiáng)制轉(zhuǎn)化，將其轉(zhuǎn)為I，并且調(diào)用channelRead0()。它是一個(gè)抽象類，因此我們?cè)诶^承SimpleChannelInboundHandler類時(shí)，需要根據(jù)自己的實(shí)際去實(shí)現(xiàn)這個(gè)方法。

四、關(guān)鍵的代碼實(shí)現(xiàn)

1、server端

（1）、SubreqServer類

package nettyserverv1; import java.util.ArrayList; import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.codec.serialization.ClassResolvers;import io.netty.handler.codec.serialization.ObjectDecoder;import io.netty.handler.codec.serialization.ObjectEncoder; /** * Created by linguolong on 2019/05/08. * Chatroom server built using netty framework */ public class SubreqServer {//保存已注冊(cè)的用戶信息public static ArrayList<UserInfo> userlist = new ArrayList<UserInfo>();//自動(dòng)生成注冊(cè)用戶static{for(int i=0;i<10;i++){UserInfo user=new UserInfo();user.setUserID('123'+i);user.setUserName('user'+i);user.setPassword('pwd'+i);userlist.add(user);}} public void bind(int port) throws Exception{ //配置服務(wù)端NIO 線程組 EventLoopGroup boss = new NioEventLoopGroup(); EventLoopGroup worker = new NioEventLoopGroup(); ServerBootstrap server = new ServerBootstrap(); try { server.group(boss, worker) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 100) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { /*** 解碼器: 構(gòu)造器傳入了兩個(gè)參數(shù): * #1 單個(gè)對(duì)象序列化后最大字節(jié)長(zhǎng)度,這是設(shè)置是1M;* #2 類解析器: weakCachingConcurrentResolver創(chuàng)建線程安全的WeakReferenceMa對(duì)類加載器進(jìn)行緩存,* 支持多線程并發(fā)訪問(wèn),當(dāng)虛擬機(jī)內(nèi)存不足時(shí),會(huì)釋放緩存中的內(nèi)存,防止內(nèi)存泄漏.*/ ch.pipeline().addLast(new ObjectDecoder(1024*1024, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())) ); ch.pipeline().addLast(new ObjectEncoder()); ch.pipeline().addLast(new SubreqServerHandler()); } }); System.out.println('Start the server success'); //綁定端口, 同步等待成功 ChannelFuture future = server.bind(port).sync(); //等待服務(wù)端監(jiān)聽(tīng)端口關(guān)閉 future.channel().closeFuture().sync(); } finally { //優(yōu)雅關(guān)閉 線程組 boss.shutdownGracefully(); worker.shutdownGracefully(); } } public static void main(String[] args) { SubreqServer server = new SubreqServer(); try { server.bind(18888); } catch (Exception e) { e.printStackTrace(); } }}

（2）、SubreqServerHandler類

package nettyserverv1; import io.netty.channel.Channel;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.channel.group.DefaultChannelGroup;import io.netty.channel.group.ChannelGroup;import io.netty.util.concurrent.GlobalEventExecutor; /** * Created by linguolong on 2019/05/08. * Chatroom client built using netty framework */ public class SubreqServerHandler extends SimpleChannelInboundHandler<String>{//新建一個(gè)channelGroup，用于存放連接的channelpublic static ChannelGroup online_channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);@Overridepublic void handlerRemoved(ChannelHandlerContext ctx){ Channel leave_channel = ctx.channel(); for (Channel channel : online_channels) { if (channel != leave_channel){ channel.writeAndFlush('[用戶 ' + leave_channel.remoteAddress() + ']下線了!n'); } } System.out.println(ctx.channel().id()+'下線了');//把剛下線的channel移除出在線用戶隊(duì)列online_channels.remove(leave_channel);} @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //判斷用戶是否已經(jīng)登錄 if(!check_login(ctx)){ //未登錄則進(jìn)行登錄驗(yàn)證 check_identity(ctx,msg); }else{ //已登錄則進(jìn)行消息轉(zhuǎn)發(fā) Channel coming_channel = ctx.channel(); for (Channel channel : online_channels) { if (channel != coming_channel){ channel.writeAndFlush('[用戶 ' + coming_channel.remoteAddress() + ']: ' + msg ); } else { channel.writeAndFlush('[我]: ' + msg); } } } } //用戶信息驗(yàn)證，檢查用戶ID和密碼是否正確 public void check_identity(ChannelHandlerContext ctx, Object msg){ UserInfo req = (UserInfo) msg; System.out.println('service receive client login req :{'+ req.toString() +'}'); boolean login_flag = false; for(int i=0;i<SubreqServer.userlist.size();i++){ if( SubreqServer.userlist.get(i).getUserID().equalsIgnoreCase(req.getUserID())&&(SubreqServer.userlist.get(i).getPassword().equals(req.getPassword()))){ login_flag=true; } } if(login_flag){ System.out.println('賬號(hào)'+req.getUserID()+'登錄成功'); ctx.writeAndFlush('您已登錄成功~n'); //將當(dāng)前的通道加入在線隊(duì)列中 online_channels.add(ctx.channel()); } else{ System.out.println('賬號(hào)'+req.getUserID()+'登錄失敗'); ctx.writeAndFlush('登錄失敗!'); //關(guān)閉連接 ctx.close(); online_channels.remove(ctx.channel()); } } //判斷用戶是否已經(jīng)在線 public boolean check_login(ChannelHandlerContext ctx){ boolean online_flag = false; for(int i=0;i<online_channels.size();i++){ if(online_channels.contains(ctx.channel())){ online_flag = true; } }return online_flag; } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { //釋放資源 ctx.close(); }@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {} //ChannelRead0()只能處理類型為String的消息，因此我們這里不能用ChannelRead0()這個(gè)方法,這里的第二個(gè)參數(shù)類型使用了泛型@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {}}

（3）、用戶信息UserInfo類

package nettyserverv1; import java.io.Serializable; /** * Created by linguolong on 2019/05/08. * Chatroom User Infomation */ public class UserInfo implements Serializable{ private String userID; private String userName; private String password; public String getUserID() { return userID; } public void setUserID(String userID) { this.userID = userID; } public String getPassword() { return password; } public void setPassword(String password) { this.password = password; } public String getUserName() { return userName; } public void setUserName(String userName) { this.userName = userName; } @Override public String toString() { return 'SubscribeReq: [messageID]:'+ userID + ' [userName]:' +userName + ' [password]:' +password; } }

2、Client端

（1）、SubreqClient類

package nettyclientv1; import java.io.BufferedReader;import java.io.InputStreamReader; import io.netty.bootstrap.Bootstrap;import io.netty.channel.Channel;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.NioSocketChannel;import io.netty.handler.codec.serialization.ClassResolvers;import io.netty.handler.codec.serialization.ObjectDecoder;import io.netty.handler.codec.serialization.ObjectEncoder; /** * Created by linguolong on 2019/05/08. * Chatroom client built using netty framework */ public class SubreqClient { public void connect(int port, String host) throws Exception{ //配置客戶端NIO 線程組 EventLoopGroup group = new NioEventLoopGroup(); Bootstrap client = new Bootstrap(); try { client.group(group) .channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { //添加解碼器 ch.pipeline().addLast(new ObjectDecoder(1024, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())) ); //添加編碼器 ch.pipeline().addLast(new ObjectEncoder()); ch.pipeline().addLast(new SubreqClientHandler()); } }); //異步獲取當(dāng)前已連接的channel Channel now_channel = client.connect(host,port).sync().channel(); BufferedReader reader = new BufferedReader(new InputStreamReader(System.in)); //異步等待客戶端連接端口關(guān)閉// now_channel.closeFuture().sync(); //使客戶端一直處于輸入狀態(tài)，直到讀取到'bye' String message = ' '; while (true) { //讀到bye時(shí)退出 if(message.equals('bye')) break; message = reader.readLine(); now_channel.writeAndFlush(message+'n'); } //讀到了'bye'字符串，主動(dòng)斷開(kāi)連接 now_channel.close(); } finally { //優(yōu)雅關(guān)閉 線程組 group.shutdownGracefully(); } } public static void main(String[] args) { SubreqClient client = new SubreqClient(); try { client.connect(18888, '127.0.0.1'); } catch (Exception e) { e.printStackTrace(); } }}

（2）、SubreqClientHandler類

package nettyclientv1; import nettyserverv1.UserInfo;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter; /** * Created by linguolong on 2019/05/08. * Chatroom client built using netty framework */ public class SubreqClientHandler extends ChannelInboundHandlerAdapter{ public SubreqClientHandler() { } /** * @param ctx * @throws Exception */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { ctx.writeAndFlush(subReq('1231','user1','pwd1')); ctx.flush(); } private UserInfo subReq(String id,String userName,String password){ UserInfo req = new UserInfo(); req.setUserID(id); req.setUserName(userName); req.setPassword(password); return req; } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { System.out.print(msg.toString()); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); }}

五、說(shuō)明

1、登錄功能的實(shí)現(xiàn)：當(dāng)客戶端剛連接上服務(wù)器時(shí)，便構(gòu)造一個(gè)UserInfo對(duì)象，對(duì)對(duì)象進(jìn)行編碼后發(fā)送給服務(wù)器。服務(wù)器接收到后對(duì)其進(jìn)行解碼，驗(yàn)證相應(yīng)的賬戶ID和密碼是否正確。

到此這篇關(guān)于java基于netty NIO的簡(jiǎn)單聊天室的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)java簡(jiǎn)單聊天室內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！