使用Netty创建一个简易的聊天室案例，不用的用户可以看到其他用户的上下线通知

代码：

服务端：

  1. package chatdemo;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.channel.ChannelFuture;
  6. import io.netty.channel.ChannelInitializer;
  7. import io.netty.channel.ChannelOption;
  8. import io.netty.channel.EventLoopGroup;
  9. import io.netty.channel.nio.NioEventLoopGroup;
  10. import io.netty.channel.socket.SocketChannel;
  11. import io.netty.channel.socket.nio.NioServerSocketChannel;
  12. import io.netty.handler.codec.string.StringDecoder;
  13. import io.netty.handler.codec.string.StringEncoder;
  15. /* 聊天室服务端 */
  16. public class GroupChatServer {
  18.     private int port;   //监听端口
  20.     public GroupChatServer(int port){
  21.         this.port = port;
  22.     }
  25.     public void run() throws Exception {
  27.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);    //管理连接请求线程组
  28.         EventLoopGroup workGroup = new NioEventLoopGroup();             //进行业务处理线程组
  29.         try {
  30.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  31.             serverBootstrap.group(bossGroup,workGroup)                      //设置两个线程组
  32.                     .channel(NioServerSocketChannel.class)                  //使用 NioSocketChannel 作为服务器通道实现
  33.                     .option(ChannelOption.SO_BACKLOG,120)             //设置线程队列等待连接的个数
  34.                     .childOption(ChannelOption.SO_KEEPALIVE,true)     //保持活动连接状态
  35.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  36.                         @Override
  37.                         protected void initChannel(SocketChannel ch) throws Exception {
  38.                             ch.pipeline().addLast("decoder",new StringDecoder());    //向PipeLine加入解码器
  39.                             ch.pipeline().addLast("encoder",new StringEncoder());    //向PipeLine加入编码器
  40.                             ch.pipeline().addLast(new GroupChatServerHandler());    //自定义业务处理Handler
  42.                         }
  43.                     });
  45.             System.out.println("netty服务器启动");
  46.             ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
  47.             channelFuture.channel().closeFuture().sync();     //对关闭通道进行监听（当有关闭通道的消息时才进行监听）
  48.         } finally {
  49.             bossGroup.shutdownGracefully();
  50.             workGroup.shutdownGracefully();
  51.         }
  53.     }
  56.     public static void main(String[] args) throws Exception {
  58.         new GroupChatServer(7000).run();
  60.     }
  63. }

服务端Handler：

  1. package chatdemo;
  3. import io.netty.channel.Channel;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.SimpleChannelInboundHandler;
  6. import io.netty.channel.group.ChannelGroup;
  7. import io.netty.channel.group.DefaultChannelGroup;
  8. import io.netty.util.concurrent.GlobalEventExecutor;
  10. import java.text.SimpleDateFormat;
  11. import java.time.LocalDateTime;
  14. /* 自定义服务端业务处理Handler*/
  15. public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {
  18.     /*定义一个Channel组，用于管理所有的Channel，当handlerRemoved方法执行时自动移除对应的Channel
  19.      * GlobalEventExecutor.INSTANCE 全局的事件执行器，是一个单例
  20.      */
  21.     private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  23.     private LocalDateTime now = LocalDateTime.now();
  26.     /* 一旦连接建立便执行该方法
  27.      *   功能: 将当前channel加入到channelGroup,
  28.      */
  29.     @Override
  30.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  32.         Channel channel = ctx.channel();
  33.         /* channelGroup的writeAndFlush方法会将自身所有的Channel进行遍历并发送消息 */
  34.         channelGroup.writeAndFlush("[客户端]"+channel.remoteAddress()+"在["+now+"]加入聊天\n");
  35.         channelGroup.add(channel);
  37.     }
  40.     /*
  41.      * 断开连接时触发
  42.      */
  43.     @Override
  44.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
  45.         Channel channel = ctx.channel();
  46.         channelGroup.writeAndFlush("[客户端]"+channel.remoteAddress()+"在["+now+"]离开了\n");
  47.         System.out.println("当前ChannelGroup容量: "+channelGroup.size());
  48.     }
  51.     /*
  52.      * 通道就绪(活动)时触发
  53.      */
  54.     @Override
  55.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  56.         System.out.println("[客户端]"+ctx.channel().remoteAddress()+"上线了!");
  57.     }
  60.     /*
  61.      * 通道非活动时触发
  62.      */
  63.     @Override
  64.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  65.         System.out.println("[客户端]"+ctx.channel().remoteAddress()+"下线了!");
  66.     }
  69.     /* 读取数据 */
  70.     @Override
  71.     protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
  73.         //获取到当前channel
  74.         Channel channel = ctx.channel();
  75.         //遍历ChannelGroup，根据不同的情况发送不同的消息
  76.         channelGroup.forEach(ch->{
  77.             if(ch!=channel){
  78.                 //不是当前Channel，进行转发
  79.                 ch.writeAndFlush("[客户]"+channel.remoteAddress()+" 发送消息 "+msg+"\n");
  80.             }else{
  81.                ch.writeAndFlush("[自己]发送了消息: "+msg+"\n");
  82.             }
  83.         });
  85.     }
  87.     /* 发生异常时触发 */
  88.     @Override
  89.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  90.         ctx.close();        //关闭通道
  91.     }
  94. }

客户端：

  1. package chatdemo;
  4. import io.netty.bootstrap.Bootstrap;
  5. import io.netty.channel.*;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  9. import io.netty.handler.codec.string.StringDecoder;
  10. import io.netty.handler.codec.string.StringEncoder;
  11. import java.util.Scanner;
  13. /* 客户端 */
  14. public class GroupChatClient {
  16.     private final String host;
  17.     private final int port;
  19.     public GroupChatClient(String host, int port) {
  20.         this.host = host;
  21.         this.port = port;
  22.     }
  24.     public void run(){
  26.         //客户端需要一个事件循环组
  27.         EventLoopGroup eventGroup = new NioEventLoopGroup();
  28.         Bootstrap bootstrap = new Bootstrap();
  29.         try {
  30.             bootstrap.group(eventGroup)
  31.                     .channel(NioSocketChannel.class)
  32.                     .handler(new ChannelInitializer<SocketChannel>() {
  33.                         @Override
  34.                         protected void initChannel(SocketChannel ch) throws Exception {
  35.                             ChannelPipeline pipeline = ch.pipeline();
  36.                             pipeline.addLast("decoder",new StringDecoder());              //向PipeLine加入解码器
  37.                             pipeline.addLast("encoder",new StringEncoder());              //向PipeLine加入编码器
  38.                             pipeline.addLast(new GroupChatClientHandler());     //添加自定义Handler
  39.                         }
  40.                     });
  41.             ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
  42.             //给关闭通道进行监听(关闭通道事件发生后触发)
  43.             Channel channel = channelFuture.channel();
  44.             System.out.println("======"+channel.localAddress()+"======");
  45.             //输入信息并发送
  46.             Scanner scanner = new Scanner(System.in);
  47.             while (scanner.hasNext()){
  48.                 String msg = scanner.nextLine();
  49.                 channel.writeAndFlush(msg+"\r\n");
  50.             }
  52.         } catch (InterruptedException e) {
  53.             e.printStackTrace();
  54.         }finally{
  55.             eventGroup.shutdownGracefully();
  56.         }
  59.     }
  63.     public static void main(String[] args) {
  64.         new GroupChatClient("127.0.0.1",7000).run();
  65.     }
  68. }

客户端Handler：

  1. package chatdemo;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.SimpleChannelInboundHandler;
  6. /* 客户端Handler */
  7. public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
  10.     @Override
  11.     protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
  12.         System.out.println(msg.trim());
  13.     }
  16. }

测试：

允许客户端多实例运行后，可以实现不同用户的群聊功能与上下线通知
image.png