一、為何要使用netty開發(fā)
由于之前已經(jīng)用Java中的socket寫過一版簡單的聊天室，這里就不再對聊天室的具體架構(gòu)進(jìn)行細(xì)致的介紹了，主要關(guān)注于使用netty框架重構(gòu)后帶來的改變。對聊天室不了解的同學(xué)可以先看下我的博客（《JAVA簡單聊天室的實現(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實現(xiàn)的，socket的處理機(jī)制屬于BIO模型，也就是阻塞IO模型。對于每一個客戶端的連接我們都需要啟動一個線程來處理，并且該線程會一直阻塞在讀取用戶數(shù)據(jù)上面。如此一來，一旦有大量的客戶端并發(fā)連接我們的服務(wù)器，服務(wù)器將難以承受。之前用JMeter測試過，單純使用Java原生socket開發(fā)的服務(wù)器所能支持的最大并發(fā)在2300左右。雖然采用線程池的策略可以在一定程度上提升最大并發(fā)數(shù)，但也無法超過1W。因此我們需要對其進(jìn)行重構(gòu)，使其能夠具有更高的性能。
2、為何使用netty框架
使用netty框架主要還是為了提升代碼的開發(fā)速度，并且減少代碼維護(hù)成本。使用netty框架開發(fā)的程序在復(fù)雜度上比使用Java原生NIO類庫開發(fā)的要小很多。具體可以看下我之前關(guān)于解決C10k問題的系列文章，里面有具體的代碼。
3、為何netty框架只實現(xiàn)了NIO而沒有AIO
前面在解決C10問題時，探究過NIO和AIO的區(qū)別，并且使用Java所提供的類庫實現(xiàn)了兩個小程序，理論上來說AIO性能明顯要比NIO高，那為什么netty使用了NIO而不是AIO呢？
官方說法如下：
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來支持AIO的實現(xiàn)，因此AIO的性能會比NIO好。而Linux上面不管是NIO還是AIO，底層都是用epoll實現(xiàn)的，性能差距不大。然而當(dāng)下絕大部分的服務(wù)器還是建立在Linux上，因此沒必要使用AIO（使用AIO反而會增加代碼的復(fù)雜度，增大維護(hù)成本）。
二、基于netty NIO的處理模型
1、服務(wù)器的類關(guān)系
（1）、SubreqServer：創(chuàng)建兩個NIO線程組，一個用來監(jiān)聽處理客戶端的連接請求，一個用來監(jiān)聽客戶端的消息。同時實例化一個ServerBootStrap啟動類的對象來啟動兩個NIO線程組，并且配置必要的參數(shù)。
（2）、ChannelInitializer：初始化SocketChannel管道的各項參數(shù)，主要有指定解碼器和編碼器，并指明管道的處理類
（3）、SubreqServerHandler：SocketChannel管道的處理類，負(fù)責(zé)處理來自客戶端的消息
2、客戶端的類關(guān)系
（1）、SubreqClient：創(chuàng)建一個NIO線程組，用來監(jiān)聽客戶端的消息。同時實例化一個ServerBootStrap啟動類的對象來啟動這個NIO線程組，并且配置必要的參數(shù)。最后讓主程序阻塞在監(jiān)聽客戶端的鍵盤輸入。
（2）、ChannelInitializer：初始化SocketChannel管道的各項參數(shù)，主要有指定解碼器和編碼器，并指明管道的處理類
（3）、SubreqClientHandler：SocketChannel管道的處理類，負(fù)責(zé)處理來自服務(wù)器的消息
三、所涉及類庫的源碼解讀
1、ChannelHandlerAdapter（用來對channel的注冊和注銷做出反應(yīng)的類）：
（1）、功能
用來實現(xiàn)當(dāng)用戶上線或下線時，通知其他在線的用戶。
（2）、類定義
它是ChannelHandler的框架實現(xiàn)
（3）、HandlerAdded()和HandlerRemoved()方法
當(dāng)有一個channel被注冊后將會調(diào)用HandlerAdded()，而當(dāng)有一個channel被注銷后將調(diào)用HandlerRemoved()方法。并且根據(jù)注釋我們知道，這兩個方法默認(rèn)不做任何處理，它希望由繼承的子類自己去寫相應(yīng)的處理實現(xiàn)。
2、SimpleChannelInboundHandler（用來對接收的消息做出反應(yīng)）
（1）、功能
用來實現(xiàn)當(dāng)接收到消息時做出相應(yīng)反應(yīng)，如果是服務(wù)端，那么將當(dāng)前消息轉(zhuǎn)發(fā)給其他在線的客戶端；如果是客戶端，就將消息簡單地打印出來。
（2）、類定義
這個類是一個泛型類，它只能用于處理一種具體類型的消息。注釋中給出一種使用方法，StringHandler繼承了SimpleChannelInboundHandler，并且指定泛型變量為String，因此這個繼承類只能用于String類型消息的處理。
（2）、channelRead(ChannelHandlerContext ctx, Object msg)
這個方法主要用來處理類型為Object的消息，也就是所有消息。
首先當(dāng)接收到msg時，先使用accpetInboundMessage()方法來判斷該消息是否可以處理。我們來看下該方法的實現(xiàn)。
如果接收到的這個消息應(yīng)該被處理就返回true，如果它應(yīng)當(dāng)被傳到ChannelPipeLine的下一個ChannelInboundHandler就返回false。
我們繼續(xù)來看下ChannelRead()方法。如果接受到的消息可以處理時。它將對消息進(jìn)行強(qiáng)制轉(zhuǎn)化，將其轉(zhuǎn)為I，并且調(diào)用channelRead0()。它是一個抽象類，因此我們在繼承SimpleChannelInboundHandler類時，需要根據(jù)自己的實際去實現(xiàn)這個方法。
四、關(guān)鍵的代碼實現(xiàn)
1、server端
（1）、SubreqServer類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
|
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 { //保存已注冊的用戶信息 public static ArrayList<UserInfo> userlist = new ArrayList<UserInfo>(); //自動生成注冊用戶 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)造器傳入了兩個參數(shù): * #1 單個對象序列化后最大字節(jié)長度,這是設(shè)置是1M; * #2 類解析器: weakCachingConcurrentResolver創(chuàng)建線程安全的WeakReferenceMa對類加載器進(jìn)行緩存, * 支持多線程并發(fā)訪問,當(dāng)虛擬機(jī)內(nèi)存不足時,會釋放緩存中的內(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)聽端口關(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類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
|
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>{ //新建一個channelGroup，用于存放連接的channel public static ChannelGroup online_channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); @Override public 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移除出在線用戶隊列 online_channels.remove(leave_channel); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //判斷用戶是否已經(jīng)登錄 if(!check_login(ctx)){ //未登錄則進(jì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); } } } } //用戶信息驗證，檢查用戶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("賬號"+req.getUserID()+"登錄成功"); ctx.writeAndFlush("您已登錄成功~\n"); //將當(dāng)前的通道加入在線隊列中 online_channels.add(ctx.channel()); } else{ System.out.println("賬號"+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(); } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { } //ChannelRead0()只能處理類型為String的消息，因此我們這里不能用ChannelRead0()這個方法,這里的第二個參數(shù)類型使用了泛型 @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { }} |
（3）、用戶信息UserInfo類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
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類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
|
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時退出 if(message.equals("bye")) break; message = reader.readLine(); now_channel.writeAndFlush(message+"\n"); } //讀到了"bye"字符串，主動斷開連接 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類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
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(); }} |
五、說明
1、登錄功能的實現(xiàn)：當(dāng)客戶端剛連接上服務(wù)器時，便構(gòu)造一個UserInfo對象，對對象進(jìn)行編碼后發(fā)送給服務(wù)器。服務(wù)器接收到后對其進(jìn)行解碼，驗證相應(yīng)的賬戶ID和密碼是否正確。
到此這篇關(guān)于java基于netty NIO的簡單聊天室的實現(xiàn)的文章就介紹到這了,更多相關(guān)java簡單聊天室內(nèi)容請搜索服務(wù)器之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持服務(wù)器之家！
原文鏈接：https://blog.csdn.net/Alexwym/article/details/90047132
