深入浅出Netty之四 Client请求处理
前2篇分析了echo server端的运行机制，本篇同样以echo client为例，分析netty的nio客户端的运行机制。
总体来说client端和server端的处理是类似的，NioWorker是重用的，也就意味着client和server的读写机制是一样的，都是通过worker线程来管理的。所不同的是Boss线程，server端的boss线程一个bind端口起一个，主要负责接收新请求，而client端的boss线程是一个可配置的数组，一个connect端口分配一个，主要负责connect过程，如果connect成功则将channle注册到worker线程中处理。在Client同样有PipelineSink，叫做NioClientSocketPipelineSink，也是负责底层IO和pipeline之间的交互。
EchoClient代码:
?
// 初始化Bootstrap和NioClientSocketChannelFactory，这一步将启动nioWorker线程，并初始化NioClientSocketPipelineSink，并将Boss线程创建 ClientBootstrap bootstrap = new ClientBootstrap( new NioClientSocketChannelFactory( Executors.newCachedThreadPool(), Executors.newCachedThreadPool())); // 用户自定义的pipeline工厂 bootstrap.setPipelineFactory(new ChannelPipelineFactory() { public ChannelPipeline getPipeline() throws Exception { return Channels.pipeline( new EchoClientHandler(firstMessageSize)); } }); // 异步创建连接 ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port)); //等待连接关闭 future.getChannel().getCloseFuture().awaitUninterruptibly(); // 关闭资源，线程池等 bootstrap.releaseExternalResources();具体connect过程：
一.创建client的channel
?
public ChannelFuture connect(final SocketAddress remoteAddress, final SocketAddress localAddress) {......//拿用户自定义的pipeline ChannelPipeline pipeline; try { pipeline = getPipelineFactory().getPipeline(); } catch (Exception e) { throw new ChannelPipelineException("Failed to initialize a pipeline.", e); } // 从ChannelFactory中，创建Channel，对于client来说factory是NioClientSocketChannelFactory，Channel是NioClientSocketChannel Channel ch = getFactory().newChannel(pipeline); // 通过channel连接 return ch.connect(remoteAddress); }?二.创建channel时，分配worker
?
public SocketChannel newChannel(ChannelPipeline pipeline) { return new NioClientSocketChannel(this, pipeline, sink, sink.nextWorker()); }?三.创建内部的SocketChannel，触发ChannelOpen事件，不过echo client的handler没有对channelOpen事件做处理
?
NioClientSocketChannel( ChannelFactory factory, ChannelPipeline pipeline, ChannelSink sink, NioWorker worker) {//创建socketChannel，并配置成异步模式 super(null, factory, pipeline, sink, newSocket(), worker); //触发open事件 fireChannelOpen(this); }?四.创建socketChannel过程
?
private static SocketChannel newSocket() { SocketChannel socket;//创建SocketChannel try { socket = SocketChannel.open(); } catch (IOException e) { throw new ChannelException("Failed to open a socket.", e); } boolean success = false;//使用异步模式 try { socket.configureBlocking(false); success = true; } catch (IOException e) { throw new ChannelException("Failed to enter non-blocking mode.", e); } finally { ...... } return socket; }?五.通过Channel进行connect
?
public static ChannelFuture connect(Channel channel, SocketAddress remoteAddress) { if (remoteAddress == null) { throw new NullPointerException("remoteAddress"); }//是否连接成功的future ChannelFuture future = future(channel, true);//触发connected的downstream事件，对于echo client来说，由于没有downstream handler，所以直接被PipelineSink处理了 channel.getPipeline().sendDownstream(new DownstreamChannelStateEvent( channel, future, ChannelState.CONNECTED, remoteAddress)); return future; }?六.NioClientSocketPipelineSink拿到CONNECTED的downstream事件，处理connect
?
case CONNECTED: if (value != null) { connect(channel, future, (SocketAddress) value); } else { channel.worker.close(channel, future); } break;private void connect( final NioClientSocketChannel channel, final ChannelFuture cf, SocketAddress remoteAddress) { try { //尝试连接，如果成功，则直接将channel注册到worker线程中 if (channel.channel.connect(remoteAddress)) { channel.worker.register(channel, cf); } //尝试连接失败，则将channel注册到某个boss线程中处理，该boss线程会接管该channel的connect过程 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)); } }七.具体register过程
void register(NioClientSocketChannel channel) { //client的注册任务，让boss线程监听connect事件 Runnable registerTask = new RegisterTask(this, channel); Selector selector; synchronized (startStopLock) { //初始化boss线程，并启动之 if (!started) { // try {//创建一个Selector this.selector = selector = Selector.open(); } catch (Throwable t) { throw new ChannelException( "Failed to create a selector.", t); } // 启动boss线程 boolean success = false; try { DeadLockProofWorker.start(bossExecutor, new ThreadRenamingRunnable(this, "New I/O client boss #" + id + '-' + subId)); success = true; } finally {...... started = true;//异步提交注册任务 boolean offered = registerTaskQueue.offer(registerTask); assert offered; } //根据超时时间，启动一个超时提醒任务 int timeout = channel.getConfig().getConnectTimeoutMillis(); if (timeout > 0) { if (!channel.isConnected()) { channel.timoutTimer = timer.newTimeout(wakeupTask, timeout, TimeUnit.MILLISECONDS); } } }八.register之后，主线程就返回了，Boss线程异步执行
?
public void run() { boolean shutdown = false; int selectReturnsImmediately = 0; Selector selector = this.selector; // use 80% of the timeout for measure final long minSelectTimeout = SelectorUtil.SELECT_TIMEOUT_NANOS * 80 / 100; boolean wakenupFromLoop = false; for (;;) { wakenUp.set(false); try { long beforeSelect = System.nanoTime(); //select，500ms超时 int selected = SelectorUtil.select(selector); ....... //处理注册任务，将channel注册到自己的selector范围中 processRegisterTaskQueue(); //处理IO就绪事件，这里是connect事件 processSelectedKeys(selector.selectedKeys()); // 处理超时，如果超时，则设置future失败 long currentTimeNanos = System.nanoTime(); processConnectTimeout(selector.keys(), currentTimeNanos); ...... }?
?九.注册任务执行
?
public void run() { try {//注册一个connect key到boss的selector上 channel.channel.register( boss.selector, SelectionKey.OP_CONNECT, channel); } catch (ClosedChannelException e) { channel.worker.close(channel, succeededFuture(channel)); } //设置超时点 int connectTimeout = channel.getConfig().getConnectTimeoutMillis(); if (connectTimeout > 0) { channel.connectDeadlineNanos = System.nanoTime() + connectTimeout * 1000000L; } }?
?十.boss处理IO，client是connect就绪
?
private void processSelectedKeys(Set<SelectionKey> selectedKeys) { ...... for (Iterator<SelectionKey> i = selectedKeys.iterator(); i.hasNext();) { SelectionKey k = i.next(); i.remove(); if (!k.isValid()) { close(k); continue; } try { //如果connect就位，判断connect是否确实完成 if (k.isConnectable()) { connect(k); } } catch (Throwable t) { ...... } } }private void connect(SelectionKey k) throws IOException { NioClientSocketChannel ch = (NioClientSocketChannel) k.attachment(); //如果连接成功，则将channel注册上worker线程中处理读写 if (ch.channel.finishConnect()) { k.cancel(); if (ch.timoutTimer != null) { ch.timoutTimer.cancel(); } ch.worker.register(ch, ch.connectFuture); } }?
?十一.注册过程
?
public void run() { ......//默认监听read事件 synchronized (channel.interestOpsLock) { channel.channel.register( selector, channel.getRawInterestOps(), channel); }//通知等待线程成功 if (future != null) { channel.setConnected(); future.setSuccess(); } ......//触发ChannelConnected事件 fireChannelConnected(channel, remoteAddress); ...... } }?十二.echo client handler处理channelConnected事件，开始发送数据
?
public void channelConnected( ChannelHandlerContext ctx, ChannelStateEvent e) { // Send the first message. Server will not send anything here // because the firstMessage's capacity is 0. e.getChannel().write(firstMessage); }?十三.write之后的过程client端和server是一样的，可以参考前一篇server端的读写
?
?