文章詳情頁

JAVA Netty實(shí)現(xiàn)聊天室+私聊功能的示例代碼

瀏覽：15日期：2022-08-26 15:53:26

功能介紹

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

服務(wù)器代碼

服務(wù)器入口代碼

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群聊服務(wù)器端 * @author zhang * */public class NettyChatServer {private int port;public NettyChatServer(int port){this.port = port;}//初始化 netty服務(wù)器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);//設(shè)置boss selector建立channel使用的對象boot.option(ChannelOption.SO_BACKLOG, 128);//boss 等待連接的隊(duì)列長度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('服務(wù)器開始啟動...');//綁定端口 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('服務(wù)器正在啟動...');}if(future.isDone()){System.out.println('服務(wù)器啟動成功...OK');}}});//監(jiān)聽channel關(guān)閉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('服務(wù)器正在關(guān)閉..');}if(future.isCancellable()){System.out.println('服務(wù)器已經(jīng)關(guān)閉..OK');}}});}finally{boss.shutdownGracefully();work.shutdownGracefully();}}/** * 啟動服務(wù)器 main 函數(shù) * @param args * @throws Exception */public static void main(String[] args) throws Exception {new NettyChatServer(9090).init();}}

服務(wù)器端消息處理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;/** * 自定義服務(wù)器端消息處理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);/** * 為了實(shí)現(xiàn)私聊功能，這里key存儲用戶的唯一標(biāo)識， * 我保存客戶端的端口號 * 當(dāng)然這個集合也需要自己去維護(hù) 用戶的上下線不能像 ChannelGroup那樣自己去維護(hù) */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();/** * 這里簡單判斷如果內(nèi)容里邊包含#那么就是私聊 */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;}//判斷當(dāng)前消息是不是自己發(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);}}}/** * 建立連接以后第一個調(diào)用的方法 */@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)就緒以后調(diào)用 */@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ā)該消息 * 同時當(dāng)前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('當(dāng)前在線人數(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('~~~~~~~~~~~~~~端口號#消息內(nèi)容~~這樣可以給單獨(dú)一個用戶發(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;/** * 客戶點(diǎn)消息處理 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();}}

測試結(jié)果

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