认识Netty

简介

应用场景

原生NIO存在的问题

Nettty的优点

Netty对JDK自带的NIO的API进行封装，解决了上述问题。

Netty版本

Netty架构设计

传统阻塞I/O模型

Reactor模式

Reactor模式基本设计思想就是I/O复用结合线程池

根据Reactor的数量和处理资源线程的数量不同，Reactor模式可以分为3种不同的实现

• 单Reactor单线程

• 单Reactor多线程

• 主从Reactor多线程

Netty模型

理论知识就整理到这，下面来个入门案例。

入门案例

• 依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.37.Final</version>
</dependency>

• 服务端

public class NettyServer {

    public static void main(String[] args) throws Exception {

        /**
         * 创建两个线程组bossGroup和workerGroup
         * bossGroup只负责连接请求, workerGroup负责处理客户端业务
         * 两个线程组包含的子线程个数: 默认是cpu核数 * 2
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //服务器启动对象
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                     .channel(NioServerSocketChannel.class)
                     .option(ChannelOption.SO_BACKLOG, 128) //设置线程队列得到的两个个数
                     .childOption(ChannelOption.SO_KEEPALIVE, true) //设置保持活动连接状态
                     .childHandler(new ChannelInitializer<SocketChannel>() {
                         protected void initChannel(SocketChannel socketChannel) throws Exception {
                             //给pipeline设置处理器
                             socketChannel.pipeline().addLast(new NettyServerHandler());
                         }
                     });
            //绑定一个端口并且同步, 生成一个ChannelFuture对象
            ChannelFuture cf = bootstrap.bind(10000).sync();
            cf.channel().closeFuture().sync();
        }finally {
            //优雅的关闭
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * @param ctx 上下文对象
     * @param msg 客户端发送的数据, 默认为Object
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("服务器线程名: "+Thread.currentThread().getName());
        System.out.println("ctx: "+ctx);
        ByteBuf buf = (ByteBuf)msg;
        System.out.println("客户端消息: "+buf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端地址: "+ctx.channel().remoteAddress());
    }

    /**
     * 数据读取完毕
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //将数据写入缓存并刷新
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 客户端!!!", CharsetUtil.UTF_8));
    }

    /**
     * 处理异常, 一般是需要关闭通道
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

• 客户端

public class NettyClient {

    public static void main(String[] args) throws Exception{
        //客户端一个事件循环组就可以了
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //启动类
            Bootstrap bootstrap = new Bootstrap();
            //设置启动参数
            bootstrap.group(group)
                    .channel(NioSocketChannel.class) //设置线程组
                    .handler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new NettyClientHandler());//加入自己的处理器
                        }
                    });
            //启动客户端连接服务器端
            ChannelFuture cf = bootstrap.connect("127.0.0.1", 10000).sync();
            cf.channel().closeFuture().sync();
        }finally {
            group.shutdownGracefully();
        }
    }
}

public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    /**
     * 通道就绪触发该方法
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello server", CharsetUtil.UTF_8));
    }

    /**
     * 当通道有读取事件时, 会触发
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf)msg;
        System.out.println("服务器：" + ctx.channel().remoteAddress() + "  "+ buf.toString(CharsetUtil.UTF_8));
    }

    /**
     * 异常处理
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

• 测试

分别启动服务端和客户端

# 服务端输出
服务器线程名: nioEventLoopGroup-3-1
ctx: ChannelHandlerContext(NettyServerHandler#0, [id: 0x2f14e16a, L:/127.0.0.1:10000 - R:/127.0.0.1:55613])
客户端消息: hello server
客户端地址: /127.0.0.1:55613

# 客户端输出
服务器：/127.0.0.1:10000  hello, 客户端!!!