JAVA Netty实现聊天室+私聊功能

功能介绍

使用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实现聊天室+私聊功能的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。