Netty剖析之Netty群聊系统

╰半橙微兮° 2023-07-03 07:21 56阅读 0赞

### 前言 ###

在前面我们讲解了Netty的基本使用，本次我们将使用Netty来完成一个群聊系统，实现服务器端和客户端之间的数据简单通讯;

群聊系统基本要求：

*  服务器端可监测客户端上线、离线、并转发客户端消息；
 *  客户端可群发消息给其它用户，也可接收其它客户发送的消息；

### 代码示例 ###

服务端：

public class GroupChatServer {
    
        private String host;
        private int port;
    
        public GroupChatServer(String host, int port) {
            this.host = host;
            this.port = port;
        }
    
        public void start() throws Exception {
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            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(new StringDecoder());
                                // 添加编码器
                                pipeline.addLast(new StringEncoder());
                                // 添加自定义业务处理handler
                                pipeline.addLast(new GroupChatServerHandler());
                            }
                        });
    
                System.out.println("Netty群聊服务器已启动~");
                ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
                channelFuture.channel().closeFuture().sync();
    
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
    
        public static void main(String[] args) throws Exception {
            new GroupChatServer("127.0.0.1", 8899).start();
        }
    }

服务端自定义handler：

public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {
    
        private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
        private static SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
        /**
         * 连接建立事件
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
    
            String msg = channel.remoteAddress() + "加入了聊天~";
            channelGroup.writeAndFlush(msg);
    
            // 将channel加入到channelGroup
            channelGroup.add(channel);
        }
    
        /**
         * 断开连接事件
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
            Channel channel = ctx.channel();
    
            String msg = "[Server]：" + channel.remoteAddress() + "下线啦~";
            channelGroup.writeAndFlush(msg);
        }
    
        /**
         * channel处理活动状态事件
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
    
            System.out.println(simpleDateFormat.format(new Date()) + "/" + ctx.channel().remoteAddress() + "上线啦~");
        }
    
        /**
         * channel处于不活动状态
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            System.out.println(simpleDateFormat.format(new Date()) + "/" + ctx.channel().remoteAddress() + " 离线了~");
        }
    
        /**
         * 读取数据事件
         *
         * @param ctx
         * @param s
         * @throws Exception
         */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
            Channel channel = ctx.channel();
            System.out.println(simpleDateFormat.format(new Date()) + "/" + "服务器接收到来自[" + channel.remoteAddress() + "]的消息，内容为[" + s + "]");
            System.out.println("服务器开始进行消息转发。。。");
            for (Channel ch : channelGroup) {
                if (channel != ch) {
                    ch.writeAndFlush(channel.remoteAddress() + "说：" + s);
                }
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

客户端：

public class GroupChatClient {
        private String serverHost;
        private int serverPort;
    
        public GroupChatClient(String serverHost, int serverPort) {
            this.serverHost = serverHost;
            this.serverPort = serverPort;
        }
    
        public void start() throws Exception {
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                ChannelPipeline pipeline = socketChannel.pipeline();
                                // 添加解码器
                                pipeline.addLast(new StringDecoder());
                                // 添加编码器
                                pipeline.addLast(new StringEncoder());
                                // 添加自定义业务处理handler
                                pipeline.addLast(new GroupChatClientHandler());
                            }
                        });
    
    
                ChannelFuture channelFuture = bootstrap.connect(serverHost, serverPort).sync();
                Channel channel = channelFuture.channel();
                System.out.println("---" + channel.localAddress() + "---");
    
                Scanner scanner = new Scanner(System.in);
                while (scanner.hasNextLine()) {
                    String msg = scanner.nextLine();
                    channel.writeAndFlush(msg);
                }
            } finally {
                group.shutdownGracefully();
            }
    
        }
    
        public static void main(String[] args) throws Exception {
    
            new GroupChatClient("127.0.0.1", 8899).start();
        }
    
    }

客户端自定义handler：

public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
            // 输出打印消息
            System.out.println(s);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

通讯演示：  
服务端：

Netty群聊服务器已启动~
    2020-02-04 12:22:11//127.0.0.1:4099上线啦~
    2020-02-04 12:22:14//127.0.0.1:4141上线啦~
    2020-02-04 12:22:24/服务器接收到来自[/127.0.0.1:4099]的消息，内容为[我是4099]
    服务器开始进行消息转发。。。
    2020-02-04 12:22:31/服务器接收到来自[/127.0.0.1:4141]的消息，内容为[我是4141]
    服务器开始进行消息转发。。。

client1：

---/127.0.0.1:4099---
    /127.0.0.1:4141加入了聊天~
    我是4099
    /127.0.0.1:4141说：我是4141

client2：

---/127.0.0.1:4141---
    /127.0.0.1:4099说：我是4099
    我是4141