当前位置:首页 > 开发 > 编程语言 > 网络编程 > 正文

netty学习之ChannelSink(NioClientSocketPipelineSink)

发表于: 2013-01-14   作者:asialee   来源:转载   浏览次数:
摘要:        这个东西应该是netty里面最难理解的,或者最关键的组件了,这个我会慢慢的进行分析。在Pipeline传送完后,都必须都通ChannelSink进行处理。Sink默认处理了琐碎的操作,例如连接、读写等等。          ChannelSink这个组件是来处理downstream请求和产生u

       这个东西应该是netty里面最难理解的,或者最关键的组件了,这个我会慢慢的进行分析。在Pipeline传送完后,都必须都通ChannelSink进行处理。Sink默认处理了琐碎的操作,例如连接、读写等等。

         ChannelSink这个组件是来处理downstream请求和产生upstream时间的一个组件,是所有io操作的执行者。也就是传输的逻辑层吧。当channel创建的时候就有一个ChannelSink和它想绑定。

        传输层的代码实现一般来说都是比较麻烦的,相比来说客户端的实现一般来说比服务端的实现要简单一些,服务端一般要处理状态变换和数据交换等,我们一点一点来看ChannelSink。

public interface ChannelSink {

    /**
     * Invoked by {@link ChannelPipeline} when a downstream {@link ChannelEvent}
     * has reached its terminal (the head of the pipeline).
     */
    void eventSunk(ChannelPipeline pipeline, ChannelEvent e) throws Exception;

    /**
     * Invoked by {@link ChannelPipeline} when an exception was raised while
     * one of its {@link ChannelHandler}s process a {@link ChannelEvent}.
     */
    void exceptionCaught(ChannelPipeline pipeline, ChannelEvent e, ChannelPipelineException cause) throws Exception;
	}

         接口里面只定义了两个操作,一个是eventSunk,一个是exceptionCaught,eventSunk我不知道怎么翻译,姑且理解为让event处理吧,一般来说应该有一个abstract骨架程序实现,我们就来看吧。

        AbstractChannelSink里面只实现了exceptionCaught,来看一眼:

        

public void exceptionCaught(ChannelPipeline pipeline,
            ChannelEvent event, ChannelPipelineException cause) throws Exception {
        Throwable actualCause = cause.getCause();
        if (actualCause == null) {
            actualCause = cause;
        }

        fireExceptionCaught(event.getChannel(), actualCause);
    }

        这个实际上是向上层报告一个DefaultExceptionEvent的upstream事件。

 

      下来我们就直奔主题来看NioClientSocketPipelineSink这个类吧。

       首先来看局部变量

private static final AtomicInteger nextId = new AtomicInteger();

    final int id = nextId.incrementAndGet();
    final Executor bossExecutor;

    private final Boss[] bosses;
    private final NioWorker[] workers;

    private final AtomicInteger bossIndex = new AtomicInteger();
    private final AtomicInteger workerIndex = new AtomicInteger();

      上面是sink的id,这个地方用到了AtomicInteger。还有Boss和NioWorker,boss的个数现在默认是1,NioWorker的个数是处理器的个数的2倍。后面两个是bossIndex和workerIndex。

     在构造函数里面初始化了Boss和NioWorker

NioClientSocketPipelineSink(
            Executor bossExecutor, Executor workerExecutor,
            int bossCount, int workerCount) {
        this.bossExecutor = bossExecutor;
        
        bosses = new Boss[bossCount];
        for (int i = 0; i < bosses.length; i ++) {
            bosses[i] = new Boss(i + 1);
        }
        
        workers = new NioWorker[workerCount];
        for (int i = 0; i < workers.length; i ++) {
            workers[i] = new NioWorker(id, i + 1, workerExecutor);
        }
    }

 我们接下来看看eventSunk的实现:

 

 public void eventSunk(
            ChannelPipeline pipeline, ChannelEvent e) throws Exception {
        if (e instanceof ChannelStateEvent) {
            ChannelStateEvent event = (ChannelStateEvent) e;
            NioClientSocketChannel channel =
                (NioClientSocketChannel) event.getChannel();
            ChannelFuture future = event.getFuture();
            ChannelState state = event.getState();
            Object value = event.getValue();

            switch (state) {
            case OPEN:
                if (Boolean.FALSE.equals(value)) {
                    channel.worker.close(channel, future);
                }
                break;
            case BOUND:
                if (value != null) {
                    bind(channel, future, (SocketAddress) value);
                } else {
                    channel.worker.close(channel, future);
                }
                break;
            case CONNECTED:
                if (value != null) {
                    connect(channel, future, (SocketAddress) value);
                } else {
                    channel.worker.close(channel, future);
                }
                break;
            case INTEREST_OPS:
                channel.worker.setInterestOps(channel, future, ((Integer) value).intValue());
                break;
            }
        } else if (e instanceof MessageEvent) {
            MessageEvent event = (MessageEvent) e;
            NioSocketChannel channel = (NioSocketChannel) event.getChannel();
            boolean offered = channel.writeBuffer.offer(event);
            assert offered;
            channel.worker.writeFromUserCode(channel);
        }
    }
         在这个里面如果是Channel的state变化的事件,则执行相应的操作,最终实现都将交给work去操作,关于worker的这些操作,我们以后会详细的介绍。

这个里面如果是bind操作的话会自己调用bind,如下:

 

private void bind(
            NioClientSocketChannel channel, ChannelFuture future,
            SocketAddress localAddress) {
        try {
            channel.socket.socket().bind(localAddress);
            channel.boundManually = true;
            channel.setBound();
            future.setSuccess();
            fireChannelBound(channel, channel.getLocalAddress());
        } catch (Throwable t) {
            future.setFailure(t);
            fireExceptionCaught(channel, t);
        }
    }

       这个里面,channel、future、localAddress都是Event里面携带的一些参数,其实这个里面实现了socket的bind操作,并触发一些事件。

最后我们看一下connect的实现:

    

private void connect(
            final NioClientSocketChannel channel, final ChannelFuture cf,
            SocketAddress remoteAddress) {
        try {
            if (channel.socket.connect(remoteAddress)) {
                channel.worker.register(channel, cf);
            } else {
                channel.getCloseFuture().addListener(new ChannelFutureListener() {
                    public void operationComplete(ChannelFuture f)
                            throws Exception {
                        if (!cf.isDone()) {
                            cf.setFailure(new ClosedChannelException());
                        }
                    }
                });
                cf.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                channel.connectFuture = cf;
                nextBoss().register(channel);
            }

        } catch (Throwable t) {
            cf.setFailure(t);
            fireExceptionCaught(channel, t);
            channel.worker.close(channel, succeededFuture(channel));
        }
    }

    这个里面会执行真正的注册动作,如果失败的话会触发一个connectFuture,这条语句 nextBoss().register(channel)不知道是什么意思?

      关于客户端channelSink就先看到这个地方,接下来会讲解Netty的线程模型,讲完之后我们再回来看服务端的channelSink的实现。

 

netty学习之ChannelSink(NioClientSocketPipelineSink)

  • 0

    开心

    开心

  • 0

    板砖

    板砖

  • 0

    感动

    感动

  • 0

    有用

    有用

  • 0

    疑问

    疑问

  • 0

    难过

    难过

  • 0

    无聊

    无聊

  • 0

    震惊

    震惊

编辑推荐
首先我们先来看serverchannel的类继承层次,然后再一个一个的研究。 现在我们来挨个的介绍各个类。
今天我来来看一下整个ChannelConfig的类的继承层次: 整个类结构来说还是比较好理解的,我就将作者的
netty里面ChannelEvent是非常重要的一个概念,在这个里面我们来重点分析一下。 ChannelEvent应该是n
首先我们先来看serverchannel的类继承层次,然后再一个一个的研究。 现在我们来挨个的介绍各个类。
今天我来来看一下整个ChannelConfig的类的继承层次: 整个类结构来说还是比较好理解的,我就将作者的
netty里面ChannelEvent是非常重要的一个概念,在这个里面我们来重点分析一下。 ChannelEvent应该是n
Netty是一个基于JAVA NIO类库的异步通信框架,它的架构特点是:异步非阻塞、基于事件驱动、高性能、
1. 背景 1.1. 惊人的性能数据 最近一个圈内朋友通过私信告诉我,通过使用Netty4 + Thrift压缩二进制
1. 背景 1.1. 惊人的性能数据 最近一个圈内朋友通过私信告诉我,通过使用Netty4 + Thrift压缩二进制
1. 背景 1.1. 惊人的性能数据 最近一个圈内朋友通过私信告诉我,通过使用Netty4 + Thrift压缩二进制
版权所有 IT知识库 CopyRight © 2009-2015 IT知识库 IT610.com , All Rights Reserved. 京ICP备09083238号