Netty浅析

2018-03-04 本文已影响23人炮灰向前冲啦

IO模型

IO多路复用模式：Reactor、Proactor

NIO实现的是Reactor模式。通过select、epoll函数，用户可以一个线程同时处理多个Channel的IO请求。当数据就绪，再通过实际的用户线程进行数据拷贝，逻辑处理

注册读事件及其对应的事件处理器
事件分离器(select\epoll)等待事件
事件到来，分离器调用相应的处理器
事件处理器完成读操作，处理数据

AIO实现的是Proactor模式。由操作系统内核负责IO数据读写，然后回调函数进行逻辑处理

事件处理器发起读请求
事件分离器等待读事件完成
在分离器等待过程中，操作系统利用并行的内核线程执行实际的读操作，并将结果数据存入用户自定义缓冲区，最后通知事件分离器读操作完成
事件分离器通知事件处理器，读操作已完成
事件处理器处理缓冲区数据

两者主要区别：用户线程或是操作系统内核线程进行IO数据读写

引入Netty

Netty中使用的Reactor模式，引入了多Reactor(1个select线程+N个IO线程+M个worker线程)。即一个主Reactor负责监控所有的连接请求，多个子Reactor负责监控并处理读/写请求，减轻了主Reactor的压力，降低了主Reactor压力太大而造成的延迟。

并且每个子Reactor分别属于一个独立的线程，每个成功连接后Channel的所有操作由同一个线程处理。这样保证了同一请求的所有状态和上下文在同一个线程中，避免了不必要的上下文切换，同时也方便了监控请求响应状态。

多Reactor模式

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

Netty服务端\客户端都需要以下两部分

至少一个ChannelHandler: 该组件实现了接收的数据处理，即消息的业务逻辑
引导Bootstrap: 服务器\客户端启动配置。比如监听端口、IO处理线程数、Channel处理逻辑

编写Echo服务端

事件处理器 ChannelHandler

@Slf4j
@ChannelHandler.Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    private final ChannelGroup channels = new DefaultChannelGroup("Echo-Server", GlobalEventExecutor.INSTANCE);

    /**
     * 客户端连接到服务端
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        log.info("channel id: {}", ctx.channel().id().asLongText());

        // 广播消息给所有channels
        channels.writeAndFlush("client: " + ctx.channel().remoteAddress() + " add");
        channels.add(ctx.channel());
    }

    /**
     * 客户端断开连接
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        channels.writeAndFlush("client: " + ctx.channel().remoteAddress() + " remove");
        channels.remove(ctx.channel());
    }

    /**
     * Channel处于活动状态，已经连接到远程节点。在线！
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("{} active", ctx.channel().remoteAddress());
    }

    /**
     * Channel未连接到远程节点。掉线！
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.warn("{} inactive", ctx.channel().remoteAddress());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf in = (ByteBuf) msg;
        log.info("Server received: {}", in.toString(StandardCharsets.UTF_8));
        ctx.writeAndFlush(in);  // write是把数据写入到OutboundBuffer(不真正发送数据)，flush是真正的发送数据
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        log.info("Server read complete.");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("Server error.");
        ctx.close();
    }
}

服务端继承ChannelInboundHandlerAdapter类，一般只需要实现channelRead()、exceptionCaught()方法即可

引导 Bootstrap

public class EchoServer {

    private final int port;

    public EchoServer(int port) {
        this.port = port;
    }

    public static void main(String[] args) {
        new EchoServer(9002).start();
    }

    public void start() {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new EchoServerHandler());
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true);

        try {
            // 阻塞绑定port，直到成功
            ChannelFuture future = bootstrap.bind(port).sync();

            // 阻塞等待，直到服务器的Channel关闭
            future.channel().closeFuture().sync();
        } catch (Exception ignore) {
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

bossGroup线程监听Channel，只需要一个，多了没用；workerGroup负责IO读写

编写Echo客户端

事件处理器 ChannelHandler

@Slf4j
@ChannelHandler.Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("Channel active.");
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", StandardCharsets.UTF_8));
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        log.info("Client received: {}", msg.toString(StandardCharsets.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("Client error.");
        ctx.close();
    }
}

客户端继承SimpleChannelInboundHandler类，此类继承ChannelInboundHandlerAdapter并实现了channelRead()方法，业务handler覆写channelRead0()方法

引导 Bootstrap

public class EchoClient {

    private final String host;
    private final int port;

    public EchoClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public static void main(String[] args) {
        new EchoClient("localhost", 9002).start();
    }

    public void start() {
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workerGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new EchoClientHandler());
                    }
                });

        try {
            ChannelFuture future = bootstrap.connect(host, port).sync();
            future.channel().closeFuture().sync();
        } catch (Exception ignore) {
        } finally {
            workerGroup.shutdownGracefully();
        }
    }
}

Netty浅析

IO模型

引入Netty

编写Echo服务端

编写Echo客户端

引申阅读

猜你喜欢

热点阅读