功能介紹

使用Netty框架實現聊天室功能，服務器可監控客戶端上下限狀態，消息轉發。同時實現了點對點私聊功能。技術點我都在代碼中做了備注，這里不再重復寫了。希望能給想學習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); //讓客戶端保持長期活動狀態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');}}});//監聽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 函數 * @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 全局事件監聽器 * 一旦將channel 加入 ChannelGroup 就不要用手動去 * 管理channel的連接失效后移除操作，他會自己移除 */private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);/** * 為了實現私聊功能，這里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;}//判斷當前消息是不是自己發送的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發送消息 * */channels.writeAndFlush(sf.format(new Date())+'n 【用戶】 '+addr+' 加入聊天室 ');channels.add(channel);String key = channel.remoteAddress().toString().split(':')[1];all.put(key, channel);}/** * channel連接狀態就緒以后調用 */@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {String addr = ctx.channel().remoteAddress().toString();System.out.println(sf.format(new Date())+' n【用戶】 '+addr+' 上線 ');}/** * channel連接狀態斷開后觸發 */@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);}/** * 連接發生異常時觸發 */@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {//System.out.println('連接發生異常！');ctx.close();}/** * 斷開連接會觸發該消息 * 同時當前channel 也會自動從ChannelGroup中被移除 */@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();String addr = channel.remoteAddress().toString();/** * 這里 ChannelGroup 底層封裝會遍歷給所有的channel發送消息 * */channels.writeAndFlush(sf.format(new Date())+'n 【用戶】 '+addr+' 離開了 ');//打印 ChannelGroup中的人數System.out.println('當前在線人數是:'+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{//創建監聽事件的監聽器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('~~~~~~~~~~~~~~端口號#消息內容~~這樣可以給單獨一個用戶發消息~~~~~~~~~~~~~~~~~~');System.out.println('#################################################');/** * 這里用控制臺輸入數據 */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);}/** * 連接異常后觸發 */@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)throws Exception {ctx.close();}}

測試結果

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

客戶端一端日志：

客戶端二日志：

客戶端三日志：

客戶端四日志：

現在在客戶端四發送消息：

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

軟化關閉客戶端客戶端三：

服務器日志：

其他客戶端日志：

發送私聊消息：

這個客戶端收不到消息

到此這篇關于JAVA Netty實現聊天室+私聊功能的示例代碼的文章就介紹到這了,更多相關JAVA Netty聊天室內容請搜索好吧啦網以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持好吧啦網！