十.Netty入门到超神系列-基于Netty群聊系统

Posted 2021-09-06 墨家巨子@俏如来

前言

本章节基于Netty做一个聊天室案例加强Netty的熟练度，案例的效果是服务端可以广播某客户端的消息给所有客户端。每个客户端监听键盘输入来获取消息，然后发送给服务端。

服务端

服务端一样的需要创建BossGroup 和 WorkGroup , 然后使用ServerBootStrap 来配置Netty和启动Netty。

public class NettyGroupChatServer {

    public static void main(String[] args) {
        new NettyGroupChatServer().start();
    }

    //监听
   public void start(){
        //循环组
       NioEventLoopGroup bossGroup = new NioEventLoopGroup();
       NioEventLoopGroup workGroup = new NioEventLoopGroup();
       //启动引导
       ServerBootstrap bootstrap = new ServerBootstrap();
       //Netty配置
       bootstrap.group(bossGroup,workGroup)
               .option(ChannelOption.SO_BACKLOG,32)
               .childOption(ChannelOption.SO_KEEPALIVE,true)
               .channel(NioserverSocketChannel.class)
               .childHandler(new ChannelInitializer<SocketChannel>() {
                   @Override
                   protected void initChannel(SocketChannel channel) throws Exception {
                       //解码器
                       channel.pipeline().addLast("decoder", new StringDecoder());
                       //编码器
                       channel.pipeline().addLast("encoder",new StringEncoder());
                       //处理器
                       channel.pipeline().addLast("nettyGroupChatHandler",new NettyGroupChatServerHandler());
                   }
               });

       try {
           //启动服务
           ChannelFuture future = bootstrap.bind(new InetSocketAddress("127.0.0.1", 8888)).sync();
           future.channel().closeFuture().sync();
       } catch (InterruptedException e) {
           e.printStackTrace();
       }finally {
           //关闭资源
           bossGroup.shutdownGracefully();
           workGroup.shutdownGracefully();
       }
   }

}

服务端处理器我们通过继承 SimpleChannelInboundHandler 入站Handler来处理消息。其中提供了几个方法

• channelRead0 ：读取消息
• handlerRemoved ：客户端断开
• handlerAdded ：客户端建立连接
• exceptionCaught ：出现异常

具体代码如下

public class NettyGroupChatServerHandler extends SimpleChannelInboundHandler<String> {

    //把所有客户端的channel保存起来
    private static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private static SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        String message = dateFormat.format(new Date())+":%s:"+msg;
        //消息转发给所有的客户端
        channels.forEach(channel -> {
            if(channel == ctx.channel()){
                String sendMsg = String.format(message, "我");
                channel.writeAndFlush(sendMsg);
                System.out.println(sendMsg);

            }else{
                String sendMsg = String.format(message, ctx.channel().remoteAddress());
                channel.writeAndFlush(sendMsg);
                System.out.println(sendMsg);
            }
        });
    }

    //断开连接 , 把消息广播给其他客户端
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        String message = dateFormat.format(new Date())+":"+ctx.channel().remoteAddress()+":断开连接";
        channels.writeAndFlush(message);
        System.out.println(message);
    }

    //建立连接
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        String message = dateFormat.format(new Date())+":"+ctx.channel().remoteAddress()+":加入聊天室";
        //自动把消息广播给其客户端
        channels.writeAndFlush(message);
        //客户端加入组
        channels.add(ctx.channel());
        System.out.println(message);
    }

    //出现异常
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.channel().close();
    }

    //客户端退出
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        String message = dateFormat.format(new Date())+":"+ctx.channel().remoteAddress()+":退出聊天室";
        System.out.println(message);
    }

    //客户端出于活动
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        String message = dateFormat.format(new Date())+":"+ctx.channel().remoteAddress()+":上线啦";
        System.out.println(message);
    }
}

客户端

客户端需要创建一个循环事件组，然后通过BootStrap去启动,然后扫描键盘输入作为数据源来把信息发送给服务端

public class NettyGroupChatClient {

    public static void main(String[] args) {
        new NettyGroupChatClient().start();
    }
    public void start(){
        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();

        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {
                        //解码器
                        channel.pipeline().addLast("decoder", new StringDecoder());
                        //编码器
                        channel.pipeline().addLast("encoder",new StringEncoder());
                        //处理器
                        channel.pipeline().addLast("nettyGroupChatClientHandler",new NettyGroupChatClientHandler());
                    }
                });
        try {
            ChannelFuture future = bootstrap.connect(new InetSocketAddress("127.0.0.1", 8888)).sync();
            //通道
            Channel channel = future.channel();
            //扫描键盘输入
            Scanner scanner = new Scanner(System.in);
            while(scanner.hasNextLine()){
                String message = scanner.nextLine();
                //发送数据
                channel.writeAndFlush(message);
            }
        } catch (InterruptedException e) {
        }finally {
            eventLoopGroup.shutdownGracefully();
        }
    }
}

对于处理器只需要接收服务端转发过来的消息即可

public class NettyGroupChatClientHandler extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(msg);
    }
}

启动服务端和多个客户端，效果如下