Netty学习开发之路

骑猪看日落 2022-06-03 07:41 244阅读 0赞

--------------------

# Netty学习 前言简介 #

=======

1.  我们公司最近决定做个基于netty tcp的IM程序，由于最近项目比较紧张，所以没办法，只能我和另外一个做安卓的同事一起学习开发。在这里先说一句瓜皮呦！而且项目分了好多个分支，分别处理不同项目的不同需求，不得不说，有一个好的PM挺重要的，然而我没遇到过。第一次写博客，比较啰嗦，希望大家别在意啊，讲的有问题的地方希望大家能都多多指正，先谢过啦！
2.  先说下这一系列博客的预期构思吧  
    1.首先肯定是netty的一些个人理解，学习的时候总发现，没有什么特别系统的netty教程，当然不是说大牛写的文章不好。主要是觉得仅仅是多功能集合实现的教程没多少。所以小弟就产生了一个大胆的想法
    
    2.其次就是netty的基础教程吧，本着帮大家度过一些坑的原则，但是要是到时候出现了一些意料之外的事情，比如用了我的办法还是报错啥的，我只能说：rua！我能怎么办，我也很绝望啊，毕竟我只是个3年的小菜鸟
    
    3.还有就是netty分布式架构的一些想法吧，最近正在实现中，目前上线的仅仅是单个netty服务器，撑tcp长连接目前不成问题。但是后期肯定得往分布式去走，然而我就被苦逼的指派过来设计了。  
    跟同事开玩笑说，拿着程序员的工资，干着架构的活 <手动捂脸>
3.  有时间我就会更新，也希望大家有啥问题能跟我多多讨论啥的。总觉得自己路子是不是走歪了，跑来写netty。无所谓了，反正本来也就不正。说下用到的东西吧  
    1.netty 4.x，不想用5.0，有些东西比较坑。  
    2.protobuf编解码，原本还是用的基础的ObjectDe/Encoder，感觉太丢人了，就换了  
    3.ChannelGroup实现广播，没用UDP，主要是怕丢包  
    4.jfinal+redis，jfinal是公司要求的，一个国产开源框架，别鄙视先，我也想用springboot，老大不让啊，j2cache是自己加的  
    5.oracle数据库，这啥都不说了，我宁愿用mysql  
    6.Quartz定时轮询程序，业务需要  
    其他的应该也没啥了吧，额…….恩，没啥了。这次就不写教程先，先贴段代码吧，省的你们说这文章水的不行  
    netty server的启动类代码，有需要的拿去改改，亲测还行。毕竟已经跑上线了，并发承载能力还不错

package cn.softsz.mc.chat.Plugin;
    
    
    import com.jfinal.log.Log;
    import com.jfinal.plugin.IPlugin;
    
    import cn.softsz.mc.chat.listener.ChatListener;
    import cn.softsz.mc.chat.pipeline.InitializerPipeline;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.PooledByteBufAllocator;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.epoll.EpollChannelOption;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    /** * netty插件的配置 * @author liuxy * */
    public class ChatPlugin implements IPlugin{ 
        private static final Log log = Log.getLog(ChatPlugin.class);
        private final EventLoopGroup bossGroup = new NioEventLoopGroup();
        private final EventLoopGroup workerGroup = new NioEventLoopGroup(4);
        private int port;
        public ChatPlugin(int port){
            this.port = port;
        }
        @Override
        public boolean start() {
            try{
                ServerBootstrap server = new ServerBootstrap();
                server.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class);
                server.childHandler(new InitializerPipeline());
                //标识当服务器请求处理线程全满时，用于临时存放已经完成三次握手的请求的队列的最大长度
                server.option(ChannelOption.SO_BACKLOG, 1024);
                server.option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_RCVBUF, 10 * 1024)  
                .option(ChannelOption.SO_SNDBUF, 10 * 1024)  
                .option(EpollChannelOption.SO_REUSEPORT, true);
                //保持长连接
                server.childOption(ChannelOption.SO_KEEPALIVE, true);
                server.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
                //通过NoDelay禁用nagle，使消息立即发出去，不用等到一定的数据量才发出去
                server.childOption(ChannelOption.TCP_NODELAY, true);
    
                Channel ch = server.bind(port).sync().channel();
                //使用监听器的方式防止启动jfinal与netty冲突
                ch.closeFuture().addListener(new ChatListener());
                log.info("SYSTEM - SERVER PORT: " + port);
                return true;
            }catch(Exception e){
                log.error("",e);
            }
            return false;
        }
    
        @Override
        public boolean stop() {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            return true;
        }

这是jfinal插件形式的，改改就能用，毕竟修修改改又是个新项目啊！

下面这个是InitializerPipeline的配置

package cn.softsz.mc.chat.pipeline;
    
    
    import java.util.concurrent.TimeUnit;
    
    import cn.softsz.mc.chat.handler.BroadcastHandler;
    import cn.softsz.mc.chat.handler.ChatHandler;
    import cn.softsz.mc.chat.handler.GroupCreateHandler;
    import cn.softsz.mc.chat.handler.LoginHandler;
    import cn.softsz.mc.chat.handler.MsgChatHandler;
    import cn.softsz.mc.chat.handler.PingHandler;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldPrepender;
    import io.netty.handler.codec.serialization.ClassResolvers;
    import io.netty.handler.codec.serialization.ObjectDecoder;
    import io.netty.handler.codec.serialization.ObjectEncoder;
    import io.netty.handler.timeout.IdleStateHandler;
    import io.netty.util.concurrent.DefaultEventExecutorGroup;
    import io.netty.util.concurrent.EventExecutorGroup;
    
    /** * 管道配置类 * @author liuxy * */
    public class InitializerPipeline extends ChannelInitializer<SocketChannel> {
        static final EventExecutorGroup MSGChATGROUP = new DefaultEventExecutorGroup(2); 
        static final EventExecutorGroup PINGChATGGROUP = new DefaultEventExecutorGroup(2); 
        static final EventExecutorGroup ChATGROUP = new DefaultEventExecutorGroup(2); 
        static final EventExecutorGroup LOGINGROUP = new DefaultEventExecutorGroup(2); 
        static final EventExecutorGroup GROUPCREATEGROUP = new DefaultEventExecutorGroup(2);
        static final EventExecutorGroup BMSGGROUP = new DefaultEventExecutorGroup(2);
        public InitializerPipeline() {
    
        }
    
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            //心跳的定时触发器
            ch.pipeline().addLast(new IdleStateHandler(41, 0, 0, TimeUnit.SECONDS)); 
            //防止TCP丢包的编解码
            pipeline.addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
            //protobuf编码器
            ch.pipeline().addLast(new ProtobufDecoder(Msg.Message.getDefaultInstance()));
            //TCP丢包解码器
            pipeline.addLast(new LengthFieldPrepender(2));
            //protobuf解码器
            pipeline.addLast(new ProtobufEncoder());
            //所有消息先到msg的handler中进行判断分发
            pipeline.addLast(MSGChATGROUP,new MsgChatHandler());
            //心跳处理器
            pipeline.addLast(PINGChATGGROUP,new PingHandler());
            //聊天信息处理器
            pipeline.addLast(ChATGROUP,new ChatHandler());
            //登陆信息处理器
            pipeline.addLast(LOGINGROUP,new LoginHandler());
            //群组创建处理器
            pipeline.addLast(GROUPCREATEGROUP,new GroupCreateHandler());
            pipeline.addLast(BMSGGROUP,new BroadcastHandler());
        }
    }

严禁抄袭！！！开玩笑的，要用的拿去，转载帮我留个地址就行。不然……..，我也没办法啊，总不能顺着网线爬过去打你吧。就先这样吧，希望大家期待我下一篇文章！略略略