netty实现消息群发

偏执的太偏执、 2023-10-12 07:56 68阅读 0赞

### netty是什么 ###

我所理解的netty，是一个比较底层的网络编程的框架，它和tomcat的区别是什么呢？就是tomcat是一个已经封装好的容器，你可以直接使用，而netty是可以写出像tomcat这种容器的。而且tomcat支持的网络协议是http，但是使用netty，可以写出支持任何协议的容易。

当然，由于所学还不够深入，暂时使用netty实现简单的功能。网上很多大神对netty的分析都很深入很到位，优点也说了很多，简单来说，使用netty可以自己搭建并实现一套高性能的服务端通讯服务，自由灵活，可扩展性好，并能支撑较好的高并发性能，国内许多知名的互联网大厂，在一些通讯组件的底层上，选用的就是netty，比如阿里的,dubbo和rocketMq，读过源码的同学应该知道就是使用netty进行封装的，所以效率才会那么高；

### 实例分析 ###

关于netty的一些基本概念和里面的一些常用组件的了解，建议可以先查阅相关资料进行学习，比如netty的编程模型，各种handler的理解，以及客户端与服务端建立连接时的channel的生命周期等概念，都是学习netty必要的基础知识；

下面有这样一个简单的需求，也是实际生活中会遇到的：用netty实现一个群聊的功能，每个用户上线和下线都能检测到，每个用户都可以发消息，每个用户发出去的消息都能被群里的其他用户接收到，下面用代码来简单实现一下，

显而易见，既然是群聊，必然有服务端和客户端，由于是后端代码，没有页面，我们将直接通过控制台发消息模拟效果即可，

服务端代码，在上一篇中我们简单分析了netty的编程特点，3个步骤，初始化服务端配置，加上一个自定义的handler即可，

**1、服务端配置**

import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /**
     * 服务端基本参数配置
     */
    public class ChatServer {
    
        private int port;
    
        public ChatServer(int port) {
            this.port = port;
        }
    
        public void startServer() throws Exception {
    
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChatServerInitializer())
                        .option(ChannelOption.SO_BACKLOG, 128)
                        .childOption(ChannelOption.SO_KEEPALIVE, true);
    
                System.out.println("Server 启动");
                ChannelFuture f = b.bind(port).sync();
                f.channel().closeFuture().sync();
            } finally {
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
                System.out.println("Server 关闭");
            }
        }
    
        public static void main(String[] args) throws Exception {
            int port;
            if (args.length > 0) {
                port = Integer.parseInt(args[0]);
            } else {
                port = 2333;
            }
            new ChatServer(port).startServer();
        }
    
    }

**2、服务端自定义ChatServerInitializer**

这部分的代码格式比较固定，可以这样理解，在客户端和服务端进行消息通讯的时候，两端都需要对进入channel的消息做一些前置的处理，比如编解码的操作，格式化输入输出校验，在netty中是通过诸如pipeline.addLast的代码实现的，也就是java中的一个责任链模式的具体实现，即把你需要的handler（处理具体某一种功能的代码块）加入到处理链中使其生效，比如这个里面最后添加了我们自定义的ChatServerHandler

public class ChatServerInitializer extends ChannelInitializer<SocketChannel> {
    
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            //以"\n"或者"\r\n"作为分隔符
            pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
            pipeline.addLast("decoder", new StringDecoder());
            pipeline.addLast("encoder", new StringEncoder());
            pipeline.addLast("handler", new ChatServerHandler());
            System.out.println("ChatClient:" + ch.remoteAddress() + "  channel_id :" + ch.id() + "  连接上");
        }
    
    }

**3、服务端自定义ChatServerHandler**  
ChatServerHandler是一个很重要的类，netty监控channel的生命周期就是在这个类里面完成的，下图可以看到，继承了SimpleChannelInboundHandler这个类之后有很多方法，在这些方法中，可以理解为netty管理具体的channel的生命周期的方法，可以根据业务的需要对这些方法进行重写，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70]

见名知意，比如说channelRegistered（）这个方法，可以理解为当channel刚注册的时候方法被触发；channelActive（）当某个客户端连接进来的时候被触发；exceptionCaught()，当channel出现异常的时候被触发，不同的回调方法都有其特定的用途，在实际生产中都是可以派上用场的，

***要特别说明的是，继承了上述的SimpleChannelInboundHandler这个类之后，有一个方法必须被复写，在4.X版本中，我这里显示的是messageReceived()，但在有一些版本中是channelRead()，但大家在学习的时候，一样的理解，里面的参数和功能是一样的，本人已经验证过了；***

理解了上述的概念，下面直接贴上handler的代码，具体的使用都有备注

import io.netty.channel.Channel;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.channel.group.ChannelGroup;
    import io.netty.channel.group.DefaultChannelGroup;
    import io.netty.util.concurrent.GlobalEventExecutor;
    
    public class ChatServerHandler extends SimpleChannelInboundHandler<String> {
    
    
        /**
         * 管理channel的组，可以理解为channel的池
         */
        public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
        // 客户端的 Channel 存入 ChannelGroup列表中，并通知列表中的其他客户端 Channel
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            Channel currchannel = ctx.channel();
            channels.writeAndFlush("[服务器] - " + currchannel.remoteAddress() + " channel_id :" + currchannel.id() + " 加入\n");
            channels.add(ctx.channel());
        }
    
        //每当从服务端收到客户端断开时，客户端的 Channel 移除 ChannelGroup 列表中，并通知列表中的其他客户端 Channel
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
            Channel currchannel = ctx.channel();
            channels.writeAndFlush("[服务器] - " + currchannel.remoteAddress() + " 离开\n");
        }
    
        //每当从服务端读到客户端写入信息时，将信息转发给其他客户端的 Channel。
        // 其中如果你使用的是 Netty 5.x 版本时，需要把 channelRead0() 重命名为messageReceived()
        @Override
        protected void messageReceived(ChannelHandlerContext channelHandlerContext, String msg) {
            Channel incoming = channelHandlerContext.channel();
            for (Channel channel : channels) {//遍历ChannelGroup中的channel
                if (channel != incoming){//找到加入到ChannelGroup中的channel后，将录入的信息回写给除去发送信息的客户端
                    channel.writeAndFlush("[" + incoming.remoteAddress() + "]" + msg + "\n");
                }
                else {
                    channel.writeAndFlush("[服务器消息]" + msg + "\n");
                }
            }
        }
    
        //服务端监听到客户端活动
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { // (5)
            Channel incoming = ctx.channel();
            System.out.println("服务器:" + incoming.remoteAddress() + "上线");
        }
    
        //服务端监听到客户端不活动
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception { // (6)
            Channel incoming = ctx.channel();
            System.out.println("服务器:" + incoming.remoteAddress() + "掉线");
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            Channel incoming = ctx.channel();
            System.out.println("服务器:" + incoming.remoteAddress() + "异常");
            ctx.close();
        }
    
    }

**4、客户端配置**  
客户端的基本连接配置和服务端差不多，不过这里的不再是ServerBootstrap，因为它是服务端专用的，这里将会变成Bootstrap，其他的都差不多，绑定端口，加入自定义的handler即可，

public class SimpleChatClient {
    
        public static void main(String[] args) throws Exception{
            new SimpleChatClient("localhost", 2333).run();
        }
    
        private final String host;
        private final int port;
    
        public SimpleChatClient(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 ChatClientInitializer());
    
                Channel channel = bootstrap.connect(host, port).sync().channel();
    
                //录入信息
                BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
                while(true){
                    //死循环，模拟不断向服务端发送消息
                    channel.writeAndFlush(in.readLine() + "\r\n");
                }
    
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                group.shutdownGracefully();
            }
        }
    }

**5、客户端ChatClientInitializer**

import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.Delimiters;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    public class ChatClientInitializer extends ChannelInitializer<SocketChannel> {
    
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            //使用netty自带的string类型的编解码器
            pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
            pipeline.addLast("decoder", new StringDecoder());
            pipeline.addLast("encoder", new StringEncoder());
            pipeline.addLast("handler", new ChatClientHandler());
        }
    }

**6、客户端自定义handler**

import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    public class ChatClientHandler extends SimpleChannelInboundHandler<String> {
    
        @Override
        protected void messageReceived(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
            //客户端读取服务端发送回来的消息，只是进行显示
            System.out.println(msg);
        }
    
    }

代码的部分基本上就是这些，下面我们运行一下程序看一下效果，首先启动服务端，  
![在这里插入图片描述][20190922105139423.png]

启动第一个客户端，客户端启动完毕，在服务端的控制台打印如下信息，说明channel的相关回调方法生效了，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 1]

再启动一个客户端，有两个客户端的连接了，同时第一个客户端收到了一条消息，即被通知另一个客户端上线了，  
![在这里插入图片描述][20190922105456899.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 2]

某个客户端的控制台输入一条消息，由于服务端是转发，因此在另一个客户端那边，收到了hello world 这条消息，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 3]

![在这里插入图片描述][20190922105619225.png]

**到这里，我们用一段简单的程序演示了netty模拟消息群发的效果，感兴趣的同学可以继续做深入的研究，本篇到此结束，最后感谢观看！**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20190922103730917.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw,size_16,color_FFFFFF,t_70
[20190922105139423.png]: https://img-blog.csdnimg.cn/20190922105139423.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2019092210523716.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw,size_16,color_FFFFFF,t_70
[20190922105456899.png]: https://img-blog.csdnimg.cn/20190922105456899.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20190922105349853.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20190922105601702.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nY29uZ3lpNDIw,size_16,color_FFFFFF,t_70
[20190922105619225.png]: https://img-blog.csdnimg.cn/20190922105619225.png