netty学习笔记10 - Netty应用实例群聊系统

桃扇骨 2023-07-06 11:36 2阅读 0赞

实例要求:

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

代码如下：

服务器

public class GroupChatServer { 
    
        // 监听端口
        private int port;
    
        public GroupChatServer(int port) { 
            this.port = port;
        }
    
        // 编写start方法，监听客户端请求
        public void run() throws InterruptedException { 
            // 创建2个线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try { 
                // -------------------------------------------------
    
                // 编写引导类，配置启动配置
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        // 配置处理类
                        .childHandler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception { 
                                // 获取到pipeline
                                ChannelPipeline pipeline = socketChannel.pipeline();
                                // 加入编解码器
                                pipeline.addLast("decoder", new StringDecoder());
                                pipeline.addLast("encoder", new StringEncoder());
                                // 加入自己的业务处理类
                                pipeline.addLast(new GroupChaServerHandler());
                            }
                        });
                System.out.println("服务器启动...");
                ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
                // 监听关闭
                channelFuture.channel().closeFuture().sync();
            } finally { 
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            GroupChatServer groupChatServer = new GroupChatServer(7000);
            groupChatServer.run();
        }
    }

服务器端处理器

public class GroupChaServerHandler extends SimpleChannelInboundHandler<String> { 
    
        // 将所有连接的客户端通道都加入到此group中
        private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
        /** *handlerAdded表示连接事件，一旦有客户端连接，第一个执行 * @param ctx * @throws Exception */
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception { 
            // 将该客户端加入聊天的信息发送给其他客户端
            Channel channel = ctx.channel();
            channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + "加入聊天");
            // 将该客户端加入到Group
            channelGroup.add(channel);
        }
    
        /** * 断开连接，有客户端断开连接，将执行此方法 * @param ctx * @throws Exception */
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { 
            // 将该客户端断开连接的信息发送给其他客户端
            Channel channel = ctx.channel();
            channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + "离开聊天");
        }
    
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception { 
            // 获取当前的channel
            Channel channel = channelHandlerContext.channel();
            // 遍历channnelGroup
            channelGroup.forEach(ch -> { 
                if(channel != ch) {  //不是当前的channel,转发消息
                    ch.writeAndFlush("[客户]" + channel.remoteAddress() + " 发送了消息" + msg + "\n");
                }else { //回显自己发送的消息给自己
                    ch.writeAndFlush("[自己]发送了消息" + msg + "\n");
                }
            });
        }
    
        //表示channel 处于活动状态, 提示 xx上线
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
    
            System.out.println(ctx.channel().remoteAddress() + " 上线了~");
        }
    
        //表示channel 处于不活动状态, 提示 xx离线了
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception { 
    
            System.out.println(ctx.channel().remoteAddress() + " 离线了~");
        }
    
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            ctx.close();
        }
    }

客户端

public class GroupChatClient { 
        //属性
        private final String host;
        private final int port;
    
        public GroupChatClient(String host, int port) { 
            this.host = host;
            this.port = port;
        }
    
        public void run() throws Exception{ 
            EventLoopGroup group = new NioEventLoopGroup();
    
            try { 
                Bootstrap bootstrap = new Bootstrap()
                        .group(group)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() { 
    
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception { 
    
                                //得到pipeline
                                ChannelPipeline pipeline = ch.pipeline();
                                //加入相关handler
                                pipeline.addLast("decoder", new StringDecoder());
                                pipeline.addLast("encoder", new StringEncoder());
                                //加入自定义的handler
                                pipeline.addLast(new GroupChatClientHandler());
                            }
                        });
    
                ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
                //得到channel
                Channel channel = channelFuture.channel();
                System.out.println("-------" + channel.localAddress()+ "--------");
                //客户端需要输入信息，创建一个扫描器
                Scanner scanner = new Scanner(System.in);
                while (scanner.hasNextLine()) { 
                    String msg = scanner.nextLine();
                    //通过channel 发送到服务器端
                    channel.writeAndFlush(msg + "\r\n");
                }
            }finally { 
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception { 
            new GroupChatClient("127.0.0.1", 7000).run();
        }
    }

客户端处理器

public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> { 
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { 
            System.out.println(msg.trim());
        }
    }