java基于netty NIO的簡單聊天室的實現(xiàn)
一、為何要使用netty開發(fā)
由于之前已經(jīng)用Java中的socket寫過一版簡單的聊天室，這里就不再對聊天室的具體架構進行細致的介紹了，主要關注于使用netty框架重構后帶來的改變。對聊天室不了解的同學可以先看下我的博客（《JAVA簡單聊天室的實現(xiàn)》）
本篇博客所使用的netty版本為4.1.36，完整工程已上傳到Github（https://github.com/Alexlingl/Chatroom），其中l(wèi)ib文件夾下有相應的netty jar包和source包，自行導入即可。
1、為何要重構
之前的聊天室是基于Java原生socket實現(xiàn)的，socket的處理機制屬于BIO模型，也就是阻塞IO模型。對于每一個客戶端的連接我們都需要啟動一個線程來處理，并且該線程會一直阻塞在讀取用戶數(shù)據(jù)上面。如此一來，一旦有大量的客戶端并發(fā)連接我們的服務器，服務器將難以承受。之前用JMeter測試過，單純使用Java原生socket開發(fā)的服務器所能支持的最大并發(fā)在2300左右。雖然采用線程池的策略可以在一定程度上提升最大并發(fā)數(shù)，但也無法超過1W。因此我們需要對其進行重構，使其能夠具有更高的性能。
2、為何使用netty框架
使用netty框架主要還是為了提升代碼的開發(fā)速度，并且減少代碼維護成本。使用netty框架開發(fā)的程序在復雜度上比使用Java原生NIO類庫開發(fā)的要小很多。具體可以看下我之前關于解決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)的，性能差距不大。然而當下絕大部分的服務器還是建立在Linux上，因此沒必要使用AIO（使用AIO反而會增加代碼的復雜度，增大維護成本）。
二、基于netty NIO的處理模型
1、服務器的類關系
（1）、SubreqServer：創(chuàng)建兩個NIO線程組，一個用來監(jiān)聽處理客戶端的連接請求，一個用來監(jiān)聽客戶端的消息。同時實例化一個ServerBootStrap啟動類的對象來啟動兩個NIO線程組，并且配置必要的參數(shù)。
（2）、ChannelInitializer：初始化SocketChannel管道的各項參數(shù)，主要有指定解碼器和編碼器，并指明管道的處理類
（3）、SubreqServerHandler：SocketChannel管道的處理類，負責處理來自客戶端的消息
2、客戶端的類關系
（1）、SubreqClient：創(chuàng)建一個NIO線程組，用來監(jiān)聽客戶端的消息。同時實例化一個ServerBootStrap啟動類的對象來啟動這個NIO線程組，并且配置必要的參數(shù)。最后讓主程序阻塞在監(jiān)聽客戶端的鍵盤輸入。
（2）、ChannelInitializer：初始化SocketChannel管道的各項參數(shù)，主要有指定解碼器和編碼器，并指明管道的處理類
（3）、SubreqClientHandler：SocketChannel管道的處理類，負責處理來自服務器的消息
三、所涉及類庫的源碼解讀
1、ChannelHandlerAdapter（用來對channel的注冊和注銷做出反應的類）：
（1）、功能
用來實現(xiàn)當用戶上線或下線時，通知其他在線的用戶。
（2）、類定義
它是ChannelHandler的框架實現(xiàn)
（3）、HandlerAdded()和HandlerRemoved()方法
當有一個channel被注冊后將會調(diào)用HandlerAdded()，而當有一個channel被注銷后將調(diào)用HandlerRemoved()方法。并且根據(jù)注釋我們知道，這兩個方法默認不做任何處理，它希望由繼承的子類自己去寫相應的處理實現(xiàn)。
2、SimpleChannelInboundHandler（用來對接收的消息做出反應）
（1）、功能
用來實現(xiàn)當接收到消息時做出相應反應，如果是服務端，那么將當前消息轉發(fā)給其他在線的客戶端；如果是客戶端，就將消息簡單地打印出來。
（2）、類定義
這個類是一個泛型類，它只能用于處理一種具體類型的消息。注釋中給出一種使用方法，StringHandler繼承了SimpleChannelInboundHandler，并且指定泛型變量為String，因此這個繼承類只能用于String類型消息的處理。
（2）、channelRead(ChannelHandlerContext ctx, Object msg)
這個方法主要用來處理類型為Object的消息，也就是所有消息。
首先當接收到msg時，先使用accpetInboundMessage()方法來判斷該消息是否可以處理。我們來看下該方法的實現(xiàn)。
如果接收到的這個消息應該被處理就返回true，如果它應當被傳到ChannelPipeLine的下一個ChannelInboundHandler就返回false。
我們繼續(xù)來看下ChannelRead()方法。如果接受到的消息可以處理時。它將對消息進行強制轉化，將其轉為I，并且調(diào)用channelRead0()。它是一個抽象類，因此我們在繼承SimpleChannelInboundHandler類時，需要根據(jù)自己的實際去實現(xiàn)這個方法。
四、關鍵的代碼實現(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 {
//保存已注冊的用戶信息
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{
//配置服務端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 {
/**
* 解碼器: 構造器傳入了兩個參數(shù):
* #1 單個對象序列化后最大字節(jié)長度,這是設置是1M;
* #2 類解析器: weakCachingConcurrentResolver創(chuàng)建線程安全的WeakReferenceMa對類加載器進行緩存,
* 支持多線程并發(fā)訪問,當虛擬機內(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();
//等待服務端監(jiān)聽端口關閉
future.channel().closeFuture().sync();
} finally {
//優(yōu)雅關閉 線程組
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>{
//新建一個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)){
//未登錄則進行登錄驗證
check_identity(ctx,msg);
}else{
//已登錄則進行消息轉發(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");
//將當前的通道加入在線隊列中
online_channels.add(ctx.channel());
}
else{
System.out.println("賬號"+req.getUserID()+"登錄失敗");
ctx.writeAndFlush("登錄失敗!");
//關閉連接
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類
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());
}
});
//異步獲取當前已連接的channel
Channel now_channel = client.connect(host,port).sync().channel();
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
//異步等待客戶端連接端口關閉
// 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)雅關閉 線程組
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();
}
}
五、說明
1、登錄功能的實現(xiàn)：當客戶端剛連接上服務器時，便構造一個UserInfo對象，對對象進行編碼后發(fā)送給服務器。服務器接收到后對其進行解碼，驗證相應的賬戶ID和密碼是否正確。
到此這篇關于java基于netty NIO的簡單聊天室的實現(xiàn)的文章就介紹到這了,更多相關java簡單聊天室內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
java動態(tài)構建數(shù)據(jù)庫復雜查詢教程
這篇文章主要介紹了java動態(tài)構建數(shù)據(jù)庫復雜查詢的實現(xiàn)示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2021-11-11
本文主要介紹了java jpa查詢沒有id表的方法,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2023-09-09
這篇文章主要介紹了java實現(xiàn)簡單石頭剪刀布游戲，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-02-02
這篇文章給大家介紹HttpClient使用，httpClient是一個客戶端的http通信實現(xiàn)庫，HttpClient的目標是發(fā)送和接收HTTP報文。本文講解的非常詳細，對HttpClient使用感興趣的朋友可以參考下2015-10-10
這篇文章主要介紹了Java實現(xiàn)基于token認證的方法示例，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-08-08
本文主要講了如何通過哈希鏈表這種數(shù)據(jù)結構來實現(xiàn)LRU算法，提供了三種實現(xiàn)思路，第一種從雙向鏈表開始，借助于HashMap來實現(xiàn)滿足要求的LRUCache2021-05-05
Spring擴展之基于HandlerMapping實現(xiàn)接口灰度發(fā)布實例
這篇文章主要介紹了Spring擴展之基于HandlerMapping實現(xiàn)接口灰度發(fā)布實例,灰度發(fā)布是指在黑與白之間,能夠平滑過渡的一種發(fā)布方式,灰度發(fā)布可以保證整體系統(tǒng)的穩(wěn)定,在初始灰度的時候就可以發(fā)現(xiàn)、調(diào)整問題,以保證其影響度,需要的朋友可以參考下2023-08-08
本文是小編給大家整理的在用java模擬http請求的時候遇到的錯誤問題整理，以及相關分析，有興趣的朋友參考下。2018-05-05
java 自己實現(xiàn)DataSource實現(xiàn)實例
這篇文章主要介紹了java 自己實現(xiàn)DataSource實現(xiàn)代碼的相關資料,需要的朋友可以參考下2017-05-05
如何用匿名內(nèi)部類實現(xiàn) Java 同步回調(diào)
這篇文章主要介紹了如何用匿名內(nèi)部類實現(xiàn) Java 同步回調(diào)，幫助大家更好的理解和學習Java，感興趣的朋友可以了解下2020-10-10