Netty学习(二)使用及执行流程(篇幅略长)

作者: 疾风老头 | 来源:发表于2019-08-21 18:26 被阅读0次

Netty简单使用

1.本文先介绍一下 server 的 demo

2.(重点是这个)根据代码跟踪一下 Netty 的一些执行流程和事件传递的 pipeline.

首先到官网看一下Netty Server 和 Client的demo, https://netty.io/wiki/user-guide-for-4.x.html, 我用的是4.1.xx,一般来说不是大版本变更, 变化不会很大.下面是 Netty Server 的demo,跟官网的是一样的.

// 下面是一个接收线程, 3个worker线程 ,所以这里是 Reactor多线程模式
// 用 Netty 的默认线程工厂,可以不传这个参数
private final static ThreadFactory threadFactory = new DefaultThreadFactory("Netty学习之路");
// Boss 线程池,用于接收客户端连接
private final static NioEventLoopGroup boss = new NioEventLoopGroup(1,threadFactory);
// Worker线程池,用于处理客户端操作
private final static NioEventLoopGroup worker = new NioEventLoopGroup(3,threadFactory);
/*
 * 下面是在构造方法中, 如果不传线程数量,默认是0, super 到 MultithreadEventLoopGroup 这里后, 最终会用 CPU核数*2 作为线程数量, Reactor多线程模式的话,就指定 boss 线程数量=1
 *  private static final int DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
 *  protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
 *  super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
 *  }
 */

public static void main(String[] args) throws Exception{
    try {
        new NettyServer(8888).start();
    }catch(Exception e){
        System.out.println("netty server启动失败");
        e.printStackTrace();
    }
}

static class NettyServer{

    private int port;

    NettyServer(int port){
        this.port = port;
    }

    void start()throws Exception{
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            ChannelFuture future = serverBootstrap
                    .group(boss, worker)
                    .channel(NioServerSocketChannel.class)
                    // 客户端连接等待队列大小
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    // 接收缓冲区
                    .option(ChannelOption.SO_RCVBUF, 32*1024)
                    // 连接超时
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10*1000)
                    .childHandler(new ChildChannelHandle())
                    .bind(this.port)
                    .sync();
            future.channel().closeFuture().sync();

        }catch(Exception e){
            throw e;
        }finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

static class ChildChannelHandle extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        // 字符串编码
        pipeline.addLast(worker,new StringEncoder());
        // 字符串解码
        pipeline.addLast(worker,new StringDecoder());
        // 自定义的handle, 状态变化后进行处理的 handle
        pipeline.addLast(worker,new StatusHandle());
        // 自定义的handle, 现在是对读取到的消息进行处理
        pipeline.addLast(worker,new CustomHandle());
    }
}

客户端的操作就简单的使用终端来操作了

image

这里对 inactive 和 active 进行了状态的输出, 输出接收数据并且原样返回给客户端

接下来看一下代码

CustomHandle

这里对接收到的客户端的数据进行处理

public class CustomHandle extends ChannelInboundHandlerAdapter {

    private Thread thread = Thread.currentThread();

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println(thread.getName()+": channelRead content : "+msg);
        ctx.writeAndFlush(msg);
    }
}

StatusHandle

对状态变化后进行处理的Handle(客户端上下线事件)

public class StatusHandle extends ChannelInboundHandlerAdapter {
    private Thread thread = Thread.currentThread();
    private String ip;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.ip = ctx.channel().remoteAddress().toString();
        System.out.println(thread.getName()+": ["+this.ip+"] channelActive -------");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(thread.getName()+": ["+this.ip+"] channelInactive -------");
    }
}

上面标记了两个地方, 从这两个地方可以窥探到 Netty 的执行流程到底是怎么样的

NioServerSocketChannel 作用相当于NIO ServerSocketChannel
ChildChannelHandle extends ChannelInitializer<SocketChannel> , 实现 initChannel 方法, 这里主要是引申出来的事件传输通道pipeline

1.NioServerSocketChannel

这个类是 Netty 用于服务端的类,用于接收客户端连接等. 用过NIO的同学都知道, serverSocket开启的时候,需要注册 ACCEPT 事件来监听客户端的连接

(小插曲)下面是Java NIO 的事件(netty基于NIO,自然也会有跟NIO一样的事件)
- public static final int OP_READ = 1 << 0; // 读消息事件
- public static final int OP_WRITE = 1 << 2; // 写消息事件
- public static final int OP_CONNECT = 1 << 3; // 连接就绪事件
- public static final int OP_ACCEPT = 1 << 4; // 新连接事件

先看一下 NioServerSocketChannel 的继承类图

image

从上面的demo的 channel(NioServerSocketChannel.class) 开始说起吧,可以看到是工厂生成channel.

public B channel(Class<? extends C> channelClass) {
    if (channelClass == null) {
        throw new NullPointerException("channelClass");
    } else {
        return this.channelFactory((io.netty.channel.ChannelFactory)(new ReflectiveChannelFactory(channelClass)));
    }
}

工厂方法生成 NioServerSocketChannel 的时候调用的构造方法:

public NioServerSocketChannel(ServerSocketChannel channel) {
    super(null, channel, SelectionKey.OP_ACCEPT);
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

继续往下跟,跟到 AbstractNioChannel 的构造方法:

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    super(parent);
    this.ch = ch;
    // 记住这个地方记录了 readInterestOp
    this.readInterestOp = readInterestOp;
    try {
            // 设置为非阻塞
        ch.configureBlocking(false);
    } catch (IOException e) {
        try {
            ch.close();
        } catch (IOException e2) {
            if (logger.isWarnEnabled()) {
                logger.warn(
                        "Failed to close a partially initialized socket.", e2);
            }
        }

        throw new ChannelException("Failed to enter non-blocking mode.", e);
    }
}

回到 ServerBootstrap 的链式调用, 接下来看 bind(port) 方法,一路追踪下去,会看到

private ChannelFuture doBind(final SocketAddress localAddress) {
// 初始化和注册
    final ChannelFuture regFuture = initAndRegister();
    final Channel channel = regFuture.channel();
    if (regFuture.cause() != null) {
        return regFuture;
    }

    if (regFuture.isDone()) {
        ChannelPromise promise = channel.newPromise();
        doBind0(regFuture, channel, localAddress, promise);
        return promise;
    } else {
        final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
        regFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                Throwable cause = future.cause();
                if (cause != null) {
                    promise.setFailure(cause);
                } else {
                    promise.registered();
                    doBind0(regFuture, channel, localAddress, promise);
                }
            }
        });
        return promise;
    }
}

看 initAndRegister 方法

final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        channel = channelFactory.newChannel();
        init(channel);
    } catch (Throwable t) {
        if (channel != null) {
            channel.unsafe().closeForcibly();
            return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
        }
        return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
    }
        // 看到这里的注册, 继续往下看
    ChannelFuture regFuture = config().group().register(channel);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }
    return regFuture;
}

config().group().register(channel); 往下看, 追踪到 AbstractChannel 的 register --> regist0(promise) (由于调用太多,省去了中间的一些调用代码)

private void register0(ChannelPromise promise) {
        try {
            // check if the channel is still open as it could be closed in the mean time when the register
            // call was outside of the eventLoop
            if (!promise.setUncancellable() || !ensureOpen(promise)) {
                return;
            }
            boolean firstRegistration = neverRegistered;
            // 执行注册
            doRegister();
            neverRegistered = false;
            registered = true;

            // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
            // user may already fire events through the pipeline in the ChannelFutureListener.
            // 这里官方也说得很清楚了,确保我们在使用 promise 的通知之前真正的调用了 pipeline 中的 handleAdded 方法
            pipeline.invokeHandlerAddedIfNeeded();

            safeSetSuccess(promise);
            // 先调用 regist 方法
            pipeline.fireChannelRegistered();
            // Only fire a channelActive if the channel has never been registered. This prevents firing
            // multiple channel actives if the channel is deregistered and re-registered.
            // 只有 channel 之前没有注册过才会调用 channelActive
            // 这里防止 channel deregistered(注销) 和 re-registered(重复调用 regist) 的时候多次调用 channelActive
            if (isActive()) {
                if (firstRegistration) {
                    // 执行 channelActive 方法
                    pipeline.fireChannelActive();
                } else if (config().isAutoRead()) {
                    // This channel was registered before and autoRead() is set. This means we need to begin read
                    // again so that we process inbound data.
                    //
                    // channel 已经注册过 并且 已经设置 autoRead().这意味着我们需要开始再次读取和处理 inbound 的数据
                    // See https://github.com/netty/netty/issues/4805
                    beginRead();
                }
            }
        } catch (Throwable t) {
            // Close the channel directly to avoid FD leak.
            closeForcibly();
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }

看到 doRegister() 方法,继续跟下去, 跟踪到 AbstractNioChannel 的 doRegister() 方法

protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
            // 这里调用java的 NIO 注册 
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                eventLoop().selectNow();
                selected = true;
            } else {
                throw e;
            }
        }
    }
}

写过NIO的同学应该熟悉上面的这句话:

selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);

这里就是调用了java NIO的注册, 至于为什么注册的时候 ops = 0
, 继续追踪下去,此处省略一堆调用....(实在是过于繁杂)最后发现, 最终都会调用 AbstractNioChannel 的 doBeginRead() 方法修改 selectionKey 的 interestOps ,客户端连接后,注册的读事件在这里也是相同的操作.

protected void doBeginRead() throws Exception {
    // Channel.read() or ChannelHandlerContext.read() was called
    final SelectionKey selectionKey = this.selectionKey;
    if (!selectionKey.isValid()) {
        return;
    }

    readPending = true;

    final int interestOps = selectionKey.interestOps();
    // // 这里是判断有没有注册过相同的事件,没有的话才修改 ops
    if ((interestOps & readInterestOp) == 0) {
        // 就是这里, 记得刚才注册的时候,ops == 0 吗, this.readInterestOp 在上面的初始化的时候赋了值
        // 与 0 逻辑或, 所以最终值就是 this.readInterestOp , 注册事件的数值 不清楚的话可以看一下最上面
        selectionKey.interestOps(interestOps | readInterestOp);
    }
}

上面介绍的服务端 ACCEPT 最后调用的 NIO 的 register 方法, read 也是调用 nio 的 register, 但是在 SocketChannel(client) 调用 register 之前, 服务端是有一个 server.accept() 方法获取客户端连接, 以此为契机, 最后我们在 NioServerSocketChannel 里面找到了accept 方法.

// 1
protected int doReadMessages(List<Object> buf) throws Exception {
    // accept 客户端, 传入 serverSocketChannel
    SocketChannel ch = SocketUtils.accept(javaChannel());

    try {
        if (ch != null) {
            // 创建新的 Netty 的 Channel , 并设置 ops =1 (read). 这是在调用 doBeginRead的时候修改的 ops 的值 , 跟 server 的一样
            buf.add(new NioSocketChannel(this, ch));
            return 1;
        }
    } catch (Throwable t) {
        logger.warn("Failed to create a new channel from an accepted socket.", t);

        try {
            ch.close();
        } catch (Throwable t2) {
            logger.warn("Failed to close a socket.", t2);
        }
    }

    return 0;
}
// 2
public static SocketChannel accept(final ServerSocketChannel serverSocketChannel) throws IOException {
    try {
        return AccessController.doPrivileged(new PrivilegedExceptionAction<SocketChannel>() {
            @Override
            public SocketChannel run() throws IOException {
                    // nio 的方法
                return serverSocketChannel.accept();
            }
        });
    } catch (PrivilegedActionException e) {
        throw (IOException) e.getCause();
    }
}

客户端连接的时候,会触发上面的 server.accept(), 然后会触发 AbstractChannel 的 register 方法从而调用下面2个方法

AbstractChannel.this.pipeline.fireChannelRegistered();// 这个方法会调用下面的两个方法

static void invokeChannelRegistered(final AbstractChannelHandlerContext next) {
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRegistered();
    } else {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                next.invokeChannelRegistered();
            }
        });
    }
}

private void invokeChannelRegistered() {
    if (invokeHandler()) {
        try {
            ((ChannelInboundHandler) handler()).channelRegistered(this);
        } catch (Throwable t) {
            notifyHandlerException(t);
        }
    } else {
        fireChannelRegistered();
    }
}

接下来我们开始讲上面提到的那个 handlerAdded 方法, 这会引申到另一个东西 pipeline.

2.ChannelInitializer

在解析这个类之前, 要先说一下 pipeline (管道,传输途径啥的都行)它就是一条 handle 消息传递链, 客户端的任何消息(事件)都经由它来处理.

先看一下 AbstractChannelHandlerContext 中的两个方法

// 查找下一个 inboundHandle (从当前位置往后查找 intBound)
private AbstractChannelHandlerContext findContextInbound() {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.next; // 往后查找
    } while (!ctx.inbound);
    return ctx;
}

// 查找下一个 OutboundHandle (从当前位置往前查找 outBound )
private AbstractChannelHandlerContext findContextOutbound() {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.prev; // 往前查找
    } while (!ctx.outbound);
    return ctx;
}

so , inbound 消息传递为从前往后, outbound 的消息传递为从后往前, 所以最先添加的 outbound 将会最后被调用

        pipeline.addLast(worker,new StringEncoder());
        // 字符串解码
        pipeline.addLast(worker,new StringDecoder());
        // 自定义的handle, 状态变化后进行处理的 handle
        pipeline.addLast(worker,new StatusHandle());
        // 自定义的handle, 现在是对读取到的消息进行处理
        pipeline.addLast(worker,new CustomHandle());

我们上面4个 handle 添加的顺序为 out, in , in, in , 所以最终调用的话,会变成下面这样

image

再看看 ChannelInitializer 这个类

public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter
/**
 * This method will be called once the {@link Channel} was registered. After the method returns this instance
 * will be removed from the {@link ChannelPipeline} of the {@link Channel}.
 *
 * @param ch            the {@link Channel} which was registered.
 * @throws Exception    is thrown if an error occurs. In that case it will be handled by
 *                      {@link #exceptionCaught(ChannelHandlerContext, Throwable)} which will by default close
 *                      the {@link Channel}.
 * 上面的意思是说,当 channel(客户端通道)一旦被注册,将会调用这个方法, 并且在方法返回的时候, 这个实例(ChannelInitializer)将会被从 ChannelPipeline (客户端的 pipeline) 中移除
 */
protected abstract void initChannel(C ch) throws Exception;

// 第一步
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
if (ctx.channel().isRegistered()) {
    initChannel(ctx);
}

// 除了这个抽象方法, 这个类还有一个重载方法
private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
    if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { // Guard against re-entrance.
        try {
            // 第二步
            // 这里调用我们自己实现的那个抽象方法 , 将 我们前面定义的 handle 都加入到 client 的 pipeline 中
            initChannel((C) ctx.channel());
        } catch (Throwable cause) {
            exceptionCaught(ctx, cause);
        } finally {
            //   第三步
            remove(ctx);
        }
        return true;
    }
    return false;
}

private void remove(ChannelHandlerContext ctx) {
    try {
        ChannelPipeline pipeline = ctx.pipeline();
        if (pipeline.context(this) != null) {
            pipeline.remove(this);
        }
    } finally {
        initMap.remove(ctx);
    }
}

Netty学习(二)使用及执行流程(篇幅略长)

Netty简单使用

1.本文先介绍一下 server 的 demo

2.(重点是这个)根据代码跟踪一下 Netty 的一些执行流程和事件传递的 pipeline.

首先到官网看一下Netty Server 和 Client的demo, https://netty.io/wiki/user-guide-for-4.x.html, 我用的是4.1.xx,一般来说不是大版本变更, 变化不会很大.下面是 Netty Server 的demo,跟官网的是一样的.

客户端的操作就简单的使用终端来操作了

这里对 inactive 和 active 进行了状态的输出, 输出接收数据并且原样返回给客户端

接下来看一下代码

CustomHandle

StatusHandle

上面标记了两个地方, 从这两个地方可以窥探到 Netty 的执行流程到底是怎么样的

NioServerSocketChannel 作用相当于NIO ServerSocketChannel

ChildChannelHandle extends ChannelInitializer<SocketChannel> , 实现 initChannel 方法, 这里主要是引申出来的事件传输通道pipeline

1.NioServerSocketChannel

写过NIO的同学应该熟悉上面的这句话:

接下来我们开始讲上面提到的那个 handlerAdded 方法, 这会引申到另一个东西 pipeline.

2.ChannelInitializer

在解析这个类之前, 要先说一下 pipeline (管道,传输途径啥的都行)它就是一条 handle 消息传递链, 客户端的任何消息(事件)都经由它来处理.

so , inbound 消息传递为从前往后, outbound 的消息传递为从后往前, 所以最先添加的 outbound 将会最后被调用

再看看 ChannelInitializer 这个类

终于写完了这一篇, 这篇的代码有点多, 如果只是demo的话, 不需要花费什么时间, 如果想要深入了解一下 Netty 的话, 可以从这里开始对源码的一点点分析.

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Netty学习(二)使用及执行流程(篇幅略长)

Netty简单使用

1.本文先介绍一下 server 的 demo

2.(重点是这个)根据代码跟踪一下 Netty 的一些执行流程 和 事件传递的 pipeline.

首先到官网看一下Netty Server 和 Client的demo, https://netty.io/wiki/user-guide-for-4.x.html, 我用的是4.1.xx,一般来说不是大版本变更, 变化不会很大.下面是 Netty Server 的demo,跟官网的是一样的.

客户端的操作就简单的使用终端来操作了

这里对 inactive 和 active 进行了状态的输出, 输出接收数据并且原样返回给客户端

接下来看一下代码

CustomHandle

StatusHandle

上面标记了两个地方, 从这两个地方可以窥探到 Netty 的执行流程到底是怎么样的

NioServerSocketChannel 作用相当于NIO ServerSocketChannel

ChildChannelHandle extends ChannelInitializer<SocketChannel> , 实现 initChannel 方法, 这里主要是引申出来的 事件传输通道pipeline

1.NioServerSocketChannel

写过NIO的同学应该熟悉上面的这句话:

接下来我们开始讲上面提到的那个 handlerAdded 方法, 这会引申到另一个东西 pipeline.

2.ChannelInitializer

在解析这个类之前, 要先说一下 pipeline (管道,传输途径啥的都行)它就是一条 handle 消息传递链, 客户端的任何消息(事件)都经由它来处理.

so , inbound 消息传递为从前往后, outbound 的消息传递为从后往前, 所以最先添加的 outbound 将会最后被调用

再看看 ChannelInitializer 这个类

终于写完了这一篇, 这篇的代码有点多, 如果只是demo的话, 不需要花费什么时间, 如果想要深入了解一下 Netty 的话, 可以从这里开始对源码的一点点分析.

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

2.(重点是这个)根据代码跟踪一下 Netty 的一些执行流程和事件传递的 pipeline.

ChildChannelHandle extends ChannelInitializer<SocketChannel> , 实现 initChannel 方法, 这里主要是引申出来的事件传输通道pipeline