功能介紹

使用Netty框架實現(xiàn)聊天室功能，服務器可監(jiān)控客戶端上下限狀態(tài)，消息轉發(fā)。同時實現(xiàn)了點對點私聊功能。技術點我都在代碼中做了備注，這里不再重復寫了。希望能給想學習netty的同學一點參考。

服務器代碼

服務器入口代碼

package nio.test.netty.groupChat;import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ChannelPipeline; 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; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.GenericFutureListener;/*** netty群聊 服務器端* @author zhang**/ public class NettyChatServer {private int port;public NettyChatServer(int port){this.port = port;}//初始化 netty服務器private void init() throws Exception{EventLoopGroup boss = new NioEventLoopGroup(1);EventLoopGroup work = new NioEventLoopGroup(16);try {ServerBootstrap boot = new ServerBootstrap();boot.group(boss,work);boot.channel(NioServerSocketChannel.class);//設置boss selector建立channel使用的對象boot.option(ChannelOption.SO_BACKLOG, 128);//boss 等待連接的 隊列長度boot.childOption(ChannelOption.SO_KEEPALIVE, true); //讓客戶端保持長期活動狀態(tài)boot.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {//從channel中獲取pipeline 并往里邊添加HandlerChannelPipeline pipeline = ch.pipeline();pipeline.addLast("encoder",new StringEncoder());pipeline.addLast("decoder",new StringDecoder());pipeline.addLast(new ServerMessageHandler());//自定義Handler來處理消息}});System.out.println("服務器開始啟動...");//綁定端口 ChannelFuture channelFuture = boot.bind(port).sync();channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {@Overridepublic void operationComplete(Future<? super Void> future)throws Exception {if(future.isSuccess()){System.out.println("服務器正在啟動...");}if(future.isDone()){System.out.println("服務器啟動成功...OK");}}});//監(jiān)聽channel關閉channelFuture.channel().closeFuture().sync();channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {@Overridepublic void operationComplete(Future<? super Void> future)throws Exception {if(future.isCancelled()){System.out.println("服務器正在關閉..");}if(future.isCancellable()){System.out.println("服務器已經關閉..OK");}}});}finally{boss.shutdownGracefully();work.shutdownGracefully();}}/*** 啟動服務器 main 函數(shù)* @param args* @throws Exception*/public static void main(String[] args) throws Exception {new NettyChatServer(9090).init();}}

服務器端消息處理Handler

package nio.test.netty.groupChat;import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; 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; import java.util.HashMap; import java.util.Map; /*** 自定義 服務器端消息處理Handler* @author zhang**/ public class ServerMessageHandler extends SimpleChannelInboundHandler<String>{/*** 管理全局的channel* GlobalEventExecutor.INSTANCE 全局事件監(jiān)聽器* 一旦將channel 加入 ChannelGroup 就不要用手動去* 管理channel的連接失效后移除操作，他會自己移除*/private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);/*** 為了實現(xiàn)私聊功能，這里key存儲用戶的唯一標識，* 我保存 客戶端的端口號* 當然 這個集合也需要自己去維護 用戶的上下線 不能像 ChannelGroup那樣自己去維護*/private static Map<String,Channel> all = new HashMap<String,Channel>();private SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");/*** 處理收到的消息*/@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg)throws Exception {Channel channel = ctx.channel();/*** 這里簡單判斷 如果內容里邊包含#那么就是私聊*/if(msg.contains("#")){String id = msg.split("#")[0];String body = msg.split("#")[1];Channel userChannel = all.get(id);String key = channel.remoteAddress().toString().split(":")[1];userChannel.writeAndFlush(sf.format(new Date())+"\n 【用戶】 "+key+" 說 : "+body);return;}//判斷當前消息是不是自己發(fā)送的for(Channel c : channels){String addr = c.remoteAddress().toString();if(channel !=c){c.writeAndFlush(sf.format(new Date())+"\n 【用戶】 "+addr+" 說 : "+msg);}else{c.writeAndFlush(sf.format(new Date())+"\n 【自己】 "+addr+" 說 : "+msg);}}}/*** 建立連接以后第一個調用的方法*/@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();String addr = channel.remoteAddress().toString();/*** 這里 ChannelGroup 底層封裝會遍歷給所有的channel發(fā)送消息* */channels.writeAndFlush(sf.format(new Date())+"\n 【用戶】 "+addr+" 加入聊天室 ");channels.add(channel);String key = channel.remoteAddress().toString().split(":")[1];all.put(key, channel);}/*** channel連接狀態(tài)就緒以后調用*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {String addr = ctx.channel().remoteAddress().toString();System.out.println(sf.format(new Date())+" \n【用戶】 "+addr+" 上線 ");}/*** channel連接狀態(tài)斷開后觸發(fā)*/@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {String addr = ctx.channel().remoteAddress().toString();System.out.println(sf.format(new Date())+" \n【用戶】 "+addr+" 下線 ");//下線移除String key = ctx.channel().remoteAddress().toString().split(":")[1];all.remove(key);}/*** 連接發(fā)生異常時觸發(fā)*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {//System.out.println("連接發(fā)生異常！");ctx.close();}/*** 斷開連接會觸發(fā)該消息* 同時當前channel 也會自動從ChannelGroup中被移除*/@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();String addr = channel.remoteAddress().toString();/*** 這里 ChannelGroup 底層封裝會遍歷給所有的channel發(fā)送消息* */channels.writeAndFlush(sf.format(new Date())+"\n 【用戶】 "+addr+" 離開了 ");//打印 ChannelGroup中的人數(shù)System.out.println("當前在線人數(shù)是:"+channels.size());System.out.println("all："+all.size());}}

客戶端主方法代碼

package nio.test.netty.groupChat;import io.netty.bootstrap.Bootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; import io.netty.util.concurrent.Future; import io.netty.util.concurrent.GenericFutureListener;import java.util.Scanner;public class NettyChatClient {private String ip;private int port;public NettyChatClient(String ip,int port){this.ip = ip;this.port = port;}/*** 初始化客戶*/private void init() throws Exception{//創(chuàng)建監(jiān)聽事件的監(jiān)聽器EventLoopGroup work = new NioEventLoopGroup();try {Bootstrap boot = new Bootstrap();boot.group(work);boot.channel(NioSocketChannel.class);boot.handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch)throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast("encoder",new StringEncoder());pipeline.addLast("decoder",new StringDecoder());pipeline.addLast(new ClientMessageHandler());}});ChannelFuture channelFuture = boot.connect(ip, port).sync();channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {@Overridepublic void operationComplete(Future<? super Void> future)throws Exception {if(future.isSuccess()){System.out.println("客戶端啟動中...");}if(future.isDone()){System.out.println("客戶端啟動成功...OK！");}}});System.out.println(channelFuture.channel().localAddress().toString());System.out.println("#################################################");System.out.println("~~~~~~~~~~~~~~端口號#消息內容~~這樣可以給單獨一個用戶發(fā)消息~~~~~~~~~~~~~~~~~~");System.out.println("#################################################");/*** 這里用控制臺輸入數(shù)據(jù)*/Channel channel = channelFuture.channel();//獲取channelScanner scanner = new Scanner(System.in);while(scanner.hasNextLine()){String str = scanner.nextLine();channel.writeAndFlush(str+"\n");}channelFuture.channel().closeFuture().sync();scanner.close();} finally {work.shutdownGracefully();}}/*** 主方法入口* @param args* @throws Exception*/public static void main(String[] args) throws Exception{new NettyChatClient("127.0.0.1",9090).init();}}

客戶端消息處理Handler

package nio.test.netty.groupChat;import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; /*** 客戶點消息處理 Handler* @author zhang**/ public class ClientMessageHandler extends SimpleChannelInboundHandler<String> {/*** 處理收到的消息*/@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg)throws Exception {System.out.println(msg);}/*** 連接異常后觸發(fā)*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {ctx.close();} }

測試結果

啟動了四個客戶端 服務器端日志效果如下：

客戶端一端日志：

客戶端二日志：

客戶端三日志：

客戶端四日志：

現(xiàn)在在客戶端四發(fā)送消息：

每個客戶端都可以收到消息：



軟化關閉客戶端客戶端三：
服務器日志：

其他客戶端日志：



發(fā)送私聊消息：

這個客戶端收不到消息

總結

以上是生活随笔為你收集整理的JAVA Netty实现聊天室+私聊功能的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。