一、实例要求

  1. 编写一个 Netty 群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）。
  2. 实现多人群聊。
  3. 服务器端：可以监测用户上线，离线，并实现消息转发功能。
  4. 客户端：通过 channel 可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息（有服务器转发得到）。

  5. 目的：进一步理解 Netty 非阻塞网络编程机制。

    二、代码

    GroupChatClient

    1. public class GroupChatClient {
    2.  private final String host;
    3.  private final int port;
    5.  public GroupChatClient(String host, int port) {
    6.      this.host = host;
    7.      this.port = port;
    8.  }
    10.  public void run() throws InterruptedException {
    11.      EventLoopGroup group = new NioEventLoopGroup();
    12.      try {
    14.          Bootstrap bootstrap = new Bootstrap();
    15.          bootstrap.group(group)
    16.                  .channel(NioSocketChannel.class)
    17.                  .handler(new ChannelInitializer<SocketChannel>() {
    18.                      @Override
    19.                      protected void initChannel(SocketChannel ch) throws Exception {
    20.                          ChannelPipeline pipeline = ch.pipeline();
    21.                          pipeline.addLast("decoder", new StringDecoder());
    22.                          pipeline.addLast("encoder", new StringEncoder());
    23.                          pipeline.addLast(new GroupChatClientHandler());
    24.                      }
    25.                  });
    27.          ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
    28.          // 得到channel
    29.          Channel channel = channelFuture.channel();
    30.          System.out.println("------" + channel.localAddress() + "--------------");
    31.          // 客户端需要输入信息，创建一个扫描器
    32.          Scanner scanner = new Scanner(System.in);
    33.          while (scanner.hasNextLine()) {
    34.              String msg = scanner.nextLine();
    35.              // 通过channel 发送到服务器端
    36.              channel.writeAndFlush(msg + "\r\n");
    37.          }
    38.      } finally {
    39.          group.shutdownGracefully();
    40.      }
    41.  }
    43.  public static void main(String[] args) throws InterruptedException {
    44.      new GroupChatClient("127.0.0.1", 6699).run();
    45.  }
    46. }

    GroupChatClientHandler

    1. public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
    2.  @Override
    3.  protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
    4.      System.out.println("客户端读取：" + msg);
    5.  }
    6. }

    GroupChatServer

    ```java public class GroupChatServer { private int port; // 监听端口

    GroupChatServer(int port) {

    1.  this.port = port;

    }

    // 编写 run 方法，处理客户端的请求 public void run() throws InterruptedException {

    1.  // 创建 两个线程组
    2.  NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
    3.  NioEventLoopGroup workerGroup = new NioEventLoopGroup(3);
    5.  try {
    6.      ServerBootstrap serverBootstrap = new ServerBootstrap();
    7.      serverBootstrap.group(bossGroup, workerGroup)
    8.              .channel(NioServerSocketChannel.class)
    9.              .option(ChannelOption.SO_BACKLOG, 128)
    10.              .childOption(ChannelOption.SO_KEEPALIVE, true)
    11.              .childHandler(new ChannelInitializer<SocketChannel>() {
    12.                  @Override
    13.                  protected void initChannel(SocketChannel ch) throws Exception {
    14.                      // 获取到 pipeline
    15.                      ChannelPipeline pipeline = ch.pipeline();
    16.                      // 向 pipeline 假如解码器
    17.                      pipeline.addLast("decoder", new StringDecoder());
    18.                      // 加入 编码器
    19.                      pipeline.addLast("coder", new StringEncoder());
    20.                      // 处理业务
    21.                      pipeline.addLast(new GroupChatServerHandler());
    22.                  }
    23.              });
    24.      System.out.println("netty 服务器启动成功");
    25.      ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
    26.      channelFuture.channel().closeFuture().sync();
    27.  } finally {
    28.      bossGroup.shutdownGracefully();
    29.      workerGroup.shutdownGracefully();
    30.  }

    }

    public static void main(String[] args) throws InterruptedException {

    1.  new GroupChatServer(6699).run();

    }

}

  1. <a name="Dg22U"></a>
  2. ## GroupChatServerHandler
  3. ```java
  4. public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {
  5.     // 定义一个channel组，管理所有的channel
  6. //    GlobalEventExecutor.INSTANCE 全局事件执行器，是一个单例
  7.     private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  8.     private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  9.     private static DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
  12.     /**
  13.      * 表示连接建立，一旦连接，第一个被执行
  14.      * 将当前 channel 加入到 channelGroup
  15.      *
  16.      * @param ctx
  17.      * @throws Exception
  18.      */
  19.     @Override
  20.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  21.         Channel channel = ctx.channel();
  22.         // 将该客户加入聊天的信息推送给其它在线的客户端
  23.         /**
  24.          * 该方法会将 channelGroup 中所有的 channel 遍历，并发送 消息
  25.          * 我们不需要自己遍历
  26.          */
  27.         channelGroup.writeAndFlush("[客户端]" + LocalDateTime.now().format(dtf) + " " + channel.remoteAddress() + " 加入聊天");
  28.         channelGroup.add(channel);
  29.     }
  31.     /**
  32.      * 表示 断开连接，将 xx客户离开信息推送给当前在线的客户
  33.      *
  34.      * @param ctx
  35.      * @throws Exception
  36.      */
  37.     @Override
  38.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
  39.         Channel channel = ctx.channel();
  40.         channelGroup.writeAndFlush("[客户端]" + LocalDateTime.now().format(dtf) + " " + channel.remoteAddress() + " 离开了");
  41.         System.out.println("channelGroupSize is " + channelGroup.size());
  42.     }
  44.     /**
  45.      * 表示 channel 处于活动状态，提示 xx 上线
  46.      *
  47.      * @param ctx
  48.      * @throws Exception
  49.      */
  50.     @Override
  51.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  52.         System.out.println(LocalDateTime.now().format(dtf) + " " + ctx.channel().remoteAddress() + " 上线了~");
  54.     }
  56.     /**
  57.      * 表示 channel 处于不活跃状态，提示 xx 下线
  58.      *
  59.      * @param ctx
  60.      * @throws Exception
  61.      */
  62.     @Override
  63.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  64.         System.out.println(LocalDateTime.now().format(dtf) + " " + ctx.channel().remoteAddress() + " 离线了~");
  65.     }
  67.     /**
  68.      * 读取数据并执行业务操作
  69.      *
  70.      * @param ctx
  71.      * @param msg
  72.      * @throws Exception
  73.      */
  74.     @Override
  75.     protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
  76.         // 获取到当前 channel 
  77.         Channel channel = ctx.channel();
  79.         // 遍历 channelgroup,根据不同的情况，回送不同的消息
  80.         channelGroup.forEach(ch -> {
  81.             if (channel != ch) {
  82.                 // 不是当前的 channel，转发消息 （不发给自己）
  83.                 ch.writeAndFlush("[客户]" + LocalDateTime.now().format(dtf) + " " + channel.remoteAddress() + "发送了消息:" + msg);
  84.             } else {
  85.                 // 发给自己
  86.                 ch.writeAndFlush("[自动发送了消息]" + sdf.format(new Date()) + " " + msg);
  87.             }
  88.         });
  89.     }
  91.     /**
  92.      * 异常
  93.      *
  94.      * @param ctx
  95.      * @param cause
  96.      * @throws Exception
  97.      */
  98.     @Override
  99.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  100.         System.out.println("【服务器异常】" + LocalDateTime.now().format(dtf) + " " + ctx.toString());
  101.         cause.printStackTrace();
  102.         ctx.close();
  103.     }
  104. }

三、结果展示

服务端

image.png

客户端

client 1

image.png

client 2

image.png

client 3

image.png