netty群聊系统

1设计思路：

1. 启动一个服务端，多个客户端

2. 第一个客户端启动时，会告诉服务器上线了

3. 第二个客户端启动时，告诉服务器上线，并且通知第一个启动的客户端

4. 第三个客户端启动时，告诉服务器上线，并且通知第一个和第二个启动的客户端

5. 其中一个客户端离开时，通知其它客户单端和服务端

2 代码

服务端：

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.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;/*** @Author: liubujun* @Date: 2023/2/15 10:31*/public class GroupChatServer {//监听端口private int port;public GroupChatServer(int port){this.port = port;}//编写run方法，处理客户端请求public void run() throws Exception{NioEventLoopGroup bossGroup = new NioEventLoopGroup();NioEventLoopGroup workGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup,workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,128).childOption(ChannelOption.SO_KEEPALIVE,true).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {//获取到pipelineChannelPipeline pipeline = ch.pipeline();//向pipeline加入解码器pipeline.addLast("decoder",new StringDecoder());//向pipeline加入编码器pipeline.addLast("encoder",new StringEncoder());//加入自己的业务处理handlepipeline.addLast(new GroupChatServerHandle());}});System.out.println("netty 服务器启动");ChannelFuture channelFuture = b.bind(port).sync();//监听关闭channelFuture.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {new GroupChatServer(7000).run();}
}

服务端处理器：

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.DefaultChannelProgressivePromise;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import jdk.nashorn.internal.runtime.GlobalConstants;import java.text.SimpleDateFormat;
import java.util.Date;/*** @Author: liubujun* @Date: 2023/2/15 14:49*/public class GroupChatServerHandle extends SimpleChannelInboundHandler<String> {//定义一个channel组，管理所有的channel//GlobalEventExecutor.INSTANCE 是全局事件的执行器，是一个单例private static ChannelGroup channelGroup =  new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);SimpleDateFormat sf =  new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//handlerAdded表示连接建立，一旦连接，表示第一个被执行//将当前channel加入到channelGroup@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();//将该客户加入的聊天信息推送给其他在线的客户端/*** 该方法会将channelGroup中所有的channel遍历，并发送，消息*/channelGroup.writeAndFlush("[客户端]"+channel.remoteAddress()+"加入聊天"+sf.format(new Date())+"\n");channelGroup.add(channel);}//断开连接，将xx客户离开信息推送给其他在线的客户端@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();channelGroup.writeAndFlush("[客户端]"+channel.remoteAddress()+"离开了");System.out.println("channelGroup size"+channelGroup.size());}//表示channel处于活动状态，提示xxx上线@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println(ctx.channel().remoteAddress()+"上线了~");}//表示channel处于非活动状态，提示xxx离线@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {System.out.println(ctx.channel().remoteAddress()+"离线了~");}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {//获取到当前的channelChannel channel = ctx.channel();//这时我们遍历channelGroup,根据不同的情况，回送不同的消息channelGroup.forEach(ch->{if (channel != ch ){ //不是当前的channel转发消息ch.writeAndFlush("[客户]"+channel.remoteAddress()+"发送了消息"+msg +"\n");}else {ch.writeAndFlush("[自己]发送了消息"+msg+"\n");}});}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();}
}

客户端代码:

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.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;/*** @Author: liubujun* @Date: 2023/2/15 10:31*/public class GroupChatServer {//监听端口private int port;public GroupChatServer(int port){this.port = port;}//编写run方法，处理客户端请求public void run() throws Exception{NioEventLoopGroup bossGroup = new NioEventLoopGroup();NioEventLoopGroup workGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup,workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,128).childOption(ChannelOption.SO_KEEPALIVE,true).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {//获取到pipelineChannelPipeline pipeline = ch.pipeline();//向pipeline加入解码器pipeline.addLast("decoder",new StringDecoder());//向pipeline加入编码器pipeline.addLast("encoder",new StringEncoder());//加入自己的业务处理handlepipeline.addLast(new GroupChatServerHandle());}});System.out.println("netty 服务器启动");ChannelFuture channelFuture = b.bind(port).sync();//监听关闭channelFuture.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {new GroupChatServer(7000).run();}
}

客户端处理器：

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;/*** @Author: liubujun* @Date: 2023/2/15 16:51*/public class GroupChatClientHandle extends SimpleChannelInboundHandler<String> {@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {System.out.println(msg.trim());}
}

演示：

先启动服务端：

启动第一个客户端，服务端和客服端控制台输出如下：

启动第二个客户端，服务端、客服端1、客户端2控制台输出如下：

启动第三个客户端，服务端控制台输出如下：

客户端1、客户端2、客户端3输出如下：

关闭其中一个客户端（我这边关闭的是第二个），服务端输出如下：

三个客户端打印台输出如下：

看到结果发现符合当初的设计。

注意：打开多个客户端需要在idea中配置：