一、介绍Netty用法之前，先聊一下I/O:

1）计算机中的信息交换机制

网络I/O 、本地I/O

2）IO的演变历史：

BIO、NIO、AIO

3）目前大家熟知的，哪些框架的底层通信在用Netty：

Dubbo、RocketMQ、Spring、ElasticSearch、GRPC

Spark、Hadoop、Flink

二、环境准备：


             
    
     io.netty
              
    
     netty-all
              
    
     4.1.36.Final

三、demo示范，相关API参考代码注释

server代码：

public class MyServer {    public static void main(String[] args) throws Exception {        EventLoopGroup bossGroup = new NioEventLoopGroup();        EventLoopGroup wokerGroup = new NioEventLoopGroup();        //epoll Only supported on Linux        //EventLoopGroup bossGroup = new EpollEventLoopGroup();        //EventLoopGroup wokerGroup = new EpollEventLoopGroup();        try{            ServerBootstrap serverBootstrap = new ServerBootstrap();            //在服务器端的handler()方法表示对bossGroup起作用，而childHandler表示对wokerGroup起作用            serverBootstrap.group(bossGroup,wokerGroup) //绑定线程池                    .channel(NioServerSocketChannel.class) //指定使用的channel                    .childHandler(new MyServerInitializer())//绑定客户端连接时触发的操作                    .localAddress(8899) //绑定端口                    .option(ChannelOption.SO_BACKLOG,128)//配置主线程分配的最大线程数                    .childOption(ChannelOption.SO_KEEPALIVE,true);//保持长连接            ChannelFuture channelFuture = serverBootstrap.bind().sync();//服务器异步创建绑定            channelFuture.channel().closeFuture().sync();//关闭服务器通道        }finally {            bossGroup.shutdownGracefully();// 释放线程池资源            wokerGroup.shutdownGracefully();// 释放线程池资源        }    }}public class MyServerHandler extends SimpleChannelInboundHandler
   
     {    @Override    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {        //打印出客户端地址        System.out.println(ctx.channel().remoteAddress()+", "+msg);        //TODO 业务逻辑处理 query(msg)        Student student = ServiceUtil.query(msg);        //TODO 返回结果做编码处理        IJacksonSerializer serializer= new JacksonSerializer();        byte[] serialize = serializer.serialize(student);        //TODO 处理结果返回给客户端        ctx.channel().writeAndFlush(serialize);    }    @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception {        System.out.println(ctx.channel().localAddress().toString() + " 通道已激活！");        super.channelActive(ctx);    }    @Override    public void channelInactive(ChannelHandlerContext ctx) throws Exception {        System.out.println(ctx.channel().localAddress().toString() + " 通道不活跃！");        super.channelInactive(ctx);    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {//cause.printStackTrace();        System.out.println(ctx.channel().localAddress().toString() + " 连接断开或者异常！");        ctx.close();    }}public class MyServerInitializer extends ChannelInitializer
    
      {    @Override    protected void initChannel(SocketChannel ch) throws Exception {        ChannelPipeline pipeline = ch.pipeline();        pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));        pipeline.addLast(new LengthFieldPrepender(4));        //字符串解码        // pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));        // //字符串编码        // pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));//字符串解码        pipeline.addLast(new ByteArrayEncoder());//字节数组编码        //自己定义的处理器        pipeline.addLast(new MyServerHandler());// 客户端触发操作    }}

Client代码：

public class MyClient {    public static void main(String[] args) throws Exception{        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();        //epoll Only supported on Linux        //EventLoopGroup eventLoopGroup = new EpollEventLoopGroup();        try{            Bootstrap bootstrap = new Bootstrap();            bootstrap.group(eventLoopGroup)                    .channel(NioSocketChannel.class) // 使用NioSocketChannel来作为连接用的channel类                    .remoteAddress(new InetSocketAddress("localhost",8899)) // 绑定连接端口和host信息                    .handler(new MyClientInitializer());            ChannelFuture channelFuture = bootstrap.connect().sync();// 异步连接服务器            channelFuture.channel().closeFuture().sync();//异步等待关闭连接channel        }finally {            eventLoopGroup.shutdownGracefully().sync();// 释放线程池资源        }    }}public class MyClientHandler extends SimpleChannelInboundHandler

四、Netty RPC支持的编解码组件很多，也可以自定义扩展：

RPC通信编解码，Netty框架支持很多种：

自定义编解码：

public class JacksonDecoder extends MessageToMessageDecoder
   
     {    private Class classtype;    public JacksonDecoder(Class classtype) {        this.classtype=classtype;    }    @Override    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List

五、ByteBuf介绍：

Netty框架自带的组件，以下面的一段代码为例，这一块涉及零拷贝相关知识点，本处不做详细展开，后续单独详解。

byte[] array = new byte[byteBuf.readableBytes()];    byteBuf.getBytes(0, array);

六、线程池：

NioEventLoopGroup 底层采用socket采用ServerSocketChannel,链接请求会注册到selector选择器：

EpollEventLoopGroup  只支持在Linux下使用，底层socket采用epoll,如果在windows下运行，服务的代码运行直接报“epoll Only supported on Linux” 错误：

底层根据实际JDK的版本，如果是Linux，则会采用epoll模型。

当前本地是WINDOWS：

转载地址：http://kosn.baihongyu.com/

你可能感兴趣的文章