网络编程之Netty实践(三)

女爷i 2022-02-25 13:42 225阅读 0赞

**一.介绍**

Netty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。也就是说，Netty 是一个基于NIO的客户、服务器端编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户、服务端应用。Netty相当于简化和流线化了网络应用的编程开发过程，例如：基于TCP和UDP的socket服务开发。  
    “快速”和“简单”并不用产生维护性或性能上的问题。Netty 是一个吸收了多种协议（包括FTP、SMTP、HTTP等各种二进制文本协议）的实现经验，并经过相当精心设计的项目。最终，Netty 成功的找到了一种方式，在保证易于开发的同时还保证了其应用的性能，稳定性和伸缩性。

**二.原理**

设计原理图:

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70]

**三.demo实战**

1.客户端最终处理器

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.util.CharsetUtil;
    import java.util.Date;
    /**
     *  客户端最终处理器
     */
    public class ClientChannelHandlerAdapter extends ChannelHandlerAdapter {
        /**
         *  捕获错误异常
         * @param ctx
         * @param cause
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.err.println("错误: "+cause.getMessage());
        }
    
        /**
         * 建立和服务器的连接后调用,一般在这里主动发送消息
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            //测试项服务端传递对象,服务端是肯定接收不到的
            ctx.writeAndFlush(new Date());
        }
    
        /**
         *  接受服务端响应
         *
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            /*ByteBuf buf = (ByteBuf) msg;
            System.out.println("收到服务器发的消息:"+buf.toString(CharsetUtil.UTF_8));*/
           //测试接受服务端传递过来的对象,肯定接收不到的
           System.out.println("收到服务器的消息:"+msg);
        }
    }

2.服务端最终处理器

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.util.CharsetUtil;
    import java.util.Date;
    
    /**
     *  服务端最终处理器
     *
     */
    public class ServerChannelHandlerAdapter extends ChannelHandlerAdapter {
    
        /**
         *  捕获netty服务异常
         *
         * @param ctx
         * @param cause
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    
            System.err.println("错误: "+cause.getMessage());
        }
    
        /**
         *  收/发消息
         * @param ctx 负责发送消息
         * @param msg 服务端收到消息
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            //测试传递原生对象,也就是收到msg,返回客户端msg,字符串的话是返回buffer
            System.out.println(msg);
            ChannelFuture channelFuture = ctx.writeAndFlush(msg);
            //注册关闭listener监听器
            channelFuture.addListener(ChannelFutureListener.CLOSE);
        }
    }

3.手写序列化工具类:

import java.io.ByteArrayInputStream;
    import java.io.ByteArrayOutputStream;
    import java.io.ObjectInputStream;
    import java.io.ObjectOutputStream;
    public class SerializationUtils {
        /**
         *  序列化
         *
         * @param obj
         * @return
         */
        public static byte[] serialize(Object obj)  {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            try{
                ObjectOutputStream oos = new ObjectOutputStream(baos);
                oos.writeObject(obj);
                oos.flush();
                oos.close();
            }catch (Exception e){
                e.printStackTrace();
            }
            return baos.toByteArray();
        }
        /**
         *  反序列化
         *
         * @param bytes
         * @return
         */
        public static Object deserialize(byte[] bytes){
            try{
                ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
                ObjectInputStream ois = new ObjectInputStream(bais);
                Object obj = ois.readObject();
                ois.close();
                return obj;
            }catch (Exception e){
                e.printStackTrace();
            }
            return null;
        }
    }

4.编码器:

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToMessageEncoder;
    import java.util.List;
    /**
     *  编码器 对Obejct对象进行编码
     */
    public class ObjectEncoder extends MessageToMessageEncoder<Object> {
        @Override
        protected void encode(ChannelHandlerContext channelHandlerContext, Object msg, List<Object> out) throws Exception {
    
            System.out.println("编码...");
            //进行对象序列化
            byte[] bytes = SerializationUtils.serialize(msg);
            //将编码之后的字节数据写到ByteBuf中
            ByteBuf buffer = Unpooled.buffer();
            buffer.writeBytes(bytes);
            //把所有对象都写进list集合中
            out.add(buffer);
        }
    }

5.解码器:

import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToMessageDecoder;
    import java.util.List;
    /**
     *  解码器 对ByteBuf进行解码,转换成对象
     */
    public class ObjectDecoder extends MessageToMessageDecoder<ByteBuf> {
        @Override
        protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf msg, List<Object> out) throws Exception {
            System.out.println("解码...");
            //将ByteBuf中的数据转换成字节数组
            byte[] bytes = new byte[msg.readableBytes()];
            msg.readBytes(bytes);
            //进行解码,转换成对象,写进list集合中也就是帧中
            Object obj = SerializationUtils.deserialize(bytes);
            out.add(obj);
        }
    }

6.初始化客户端通信管道:

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;
    /**
     *  初始化客户端通信管道
     */
    public class ClientchannelInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            //获取管道
            ChannelPipeline pipeline = socketChannel.pipeline();
            //数据帧头解码器                                    最大帧长度2的16次方减一        偏移量           数据帧头长度         长度调整量         数据截取跳的长度
            pipeline.addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
            //添加数据帧头编码器
            pipeline.addLast(new LengthFieldPrepender(2));
            //添加解码器
            pipeline.addLast(new ObjectDecoder());
            //添加编码器
            pipeline.addLast(new ObjectEncoder());
            //指定消息最终处理者
            pipeline.addLast(new ClientChannelHandlerAdapter());
        }
    }

7.初始化服务端通信管道:

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;
    /**
     *
     * 初始化服务端通信管道
     *
     */
    public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            //获取管道
            ChannelPipeline pipeline = socketChannel.pipeline();
            //数据帧头解码器                                    最大帧长度2的16次方减一        偏移量         数据帧头长度        长度调整量      数据截取跳的长度
            pipeline.addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
            //添加数据帧头编码器
            pipeline.addLast(new LengthFieldPrepender(2));
            //添加解码器
            pipeline.addLast(new ObjectDecoder());
            //添加编码器
            pipeline.addLast(new ObjectEncoder());
            //添加最终处理者
            pipeline.addLast(new ServerChannelHandlerAdapter());
        }
    }

**四.最终demo实践**

1.客户端

import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    /**
     *  netty 客户端
     */
    public class NettyClientDemo {
        public static void main(String[] args) throws InterruptedException {
            //1.创建客户端引导
            Bootstrap bt = new Bootstrap();
            //2.创建线程池 worker请求处理
            NioEventLoopGroup worker = new NioEventLoopGroup();
            //3.关联线程池
            bt.group(worker);
            //4.设置Nio客户端实现
            bt.channel(NioSocketChannel.class);
            //5.初始化通信管道  ----重点
            bt.handler(new ClientchannelInitializer());
            //6.连接服务端
            ChannelFuture channelFuture = bt.connect("127.0.0.1",9999).sync();
            //7.关闭与服务端之间的通信管道
            channelFuture.channel().closeFuture().sync();
            //8.释放线程资源
            worker.shutdownGracefully();
        }
    }

2.服务端

import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    /**
     *  netty服务端
     */
    public class NettyServerDemo {
        public static void main(String[] args) throws InterruptedException {
            //1.创建服务启动引导
            ServerBootstrap sbt = new ServerBootstrap();
            //2.创建线程池 boss负责请求转发 worker负责相应处理
            NioEventLoopGroup boss = new NioEventLoopGroup();
            NioEventLoopGroup worker = new NioEventLoopGroup();
            //3.关联线程池
            sbt.group(boss,worker);
            //4.设置Nio服务端实现
            sbt.channel(NioServerSocketChannel.class);
            //5.初始化通讯管道 -----重点
            sbt.childHandler(new ServerChannelInitializer());
            //6.绑定端口启动netty服务
            System.out.println("我在9999监听....");
            ChannelFuture channelFuture = sbt.bind(9999).sync();
            //7.关闭与客户端间的通讯管道
            channelFuture.channel().closeFuture().sync();
            //8.释放线程资源
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }

**五.总结**  
    以上就完成了netty的设计原理以及demo实践过程,总的来说netty是非常好用的,尤其是在通过Netty手写RPC通信的时候,以后我会更新一篇手写RPC的文章,以上就完成了从BIO–>NIO–>Netty的研究和学习。

不仅要知其依然也要知其所以然          -----李国辉

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70]: /images/20220225/63b78beaa40e45719270f639db89590c.png