Netty（四）:AbstractChannel源码解析

   1 /*
   2  * Copyright 2012 The Netty Project
   3  *
   4  * The Netty Project licenses this file to you under the Apache License,
   5  * version 2.0 (the "License"); you may not use this file except in compliance
   6  * with the License. You may obtain a copy of the License at:
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12  * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13  * License for the specific language governing permissions and limitations
  14  * under the License.
  15  */
  16 package io.netty.channel;
  17 
  18 import io.netty.buffer.ByteBufAllocator;
  19 import io.netty.channel.socket.ChannelOutputShutdownEvent;
  20 import io.netty.channel.socket.ChannelOutputShutdownException;
  21 import io.netty.util.DefaultAttributeMap;
  22 import io.netty.util.ReferenceCountUtil;
  23 import io.netty.util.internal.PlatformDependent;
  24 import io.netty.util.internal.ThrowableUtil;
  25 import io.netty.util.internal.UnstableApi;
  26 import io.netty.util.internal.logging.InternalLogger;
  27 import io.netty.util.internal.logging.InternalLoggerFactory;
  28 
  29 import java.io.IOException;
  30 import java.net.ConnectException;
  31 import java.net.InetSocketAddress;
  32 import java.net.NoRouteToHostException;
  33 import java.net.SocketAddress;
  34 import java.net.SocketException;
  35 import java.nio.channels.ClosedChannelException;
  36 import java.nio.channels.NotYetConnectedException;
  37 import java.util.concurrent.Executor;
  38 import java.util.concurrent.RejectedExecutionException;
  39 
  40 /**
  41  * A skeletal {@link Channel} implementation.
  42  */
  43 public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
  44 
  45     private static final InternalLogger logger = InternalLoggerFactory.getInstance(AbstractChannel.class);
  46 
  47     private static final ClosedChannelException FLUSH0_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
  48             new ClosedChannelException(), AbstractUnsafe.class, "flush0()");
  49     private static final ClosedChannelException ENSURE_OPEN_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
  50             new ClosedChannelException(), AbstractUnsafe.class, "ensureOpen(...)");
  51     private static final ClosedChannelException CLOSE_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
  52             new ClosedChannelException(), AbstractUnsafe.class, "close(...)");
  53     private static final ClosedChannelException WRITE_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
  54             new ClosedChannelException(), AbstractUnsafe.class, "write(...)");
  55     private static final NotYetConnectedException FLUSH0_NOT_YET_CONNECTED_EXCEPTION = ThrowableUtil.unknownStackTrace(
  56             new NotYetConnectedException(), AbstractUnsafe.class, "flush0()");
  57 
  58     //父Channel
  59     private final Channel parent;
  60     private final ChannelId id;
  61     //Unsafe对象，封装ByteBuf的读写过程
  62     private final Unsafe unsafe;
  63     //关联的Pipeline对象
  64     private final DefaultChannelPipeline pipeline;
  65 
  66     private final VoidChannelPromise unsafeVoidPromise = new VoidChannelPromise(this, false);
  67     private final CloseFuture closeFuture = new CloseFuture(this);
  68 
  69     //本地地址和远端地址
  70     private volatile SocketAddress localAddress;
  71     private volatile SocketAddress remoteAddress;
  72 
  73     //用来处理读写的线程
  74     private volatile EventLoop eventLoop;
  75 
  76     private volatile boolean registered;
  77     private boolean closeInitiated;
  78 
  79     /** Cache for the string representation of this channel */
  80     private boolean strValActive;
  81     private String strVal;
  82 
  83     /**
  84      * Creates a new instance.
  85      *
  86      * @param parent
  87      *        the parent of this channel. {@code null} if there‘s no parent.
  88      */
  89     protected AbstractChannel(Channel parent) {
  90         this.parent = parent;
  91         id = newId();
  92         unsafe = newUnsafe();
  93         pipeline = newChannelPipeline();
  94     }
  95 
  96     /**
  97      * Creates a new instance.
  98      *
  99      * @param parent
 100      *        the parent of this channel. {@code null} if there‘s no parent.
 101      */
 102     protected AbstractChannel(Channel parent, ChannelId id) {
 103         this.parent = parent;
 104         this.id = id;
 105         unsafe = newUnsafe();
 106         pipeline = newChannelPipeline();
 107     }
 108 
 109     @Override
 110     public final ChannelId id() {
 111         return id;
 112     }
 113 
 114     /**
 115      * Returns a new {@link DefaultChannelId} instance. Subclasses may override this method to assign custom
 116      * {@link ChannelId}s to {@link Channel}s that use the {@link AbstractChannel#AbstractChannel(Channel)} constructor.
 117      */
 118     protected ChannelId newId() {
 119         return DefaultChannelId.newInstance();
 120     }
 121 
 122     /**
 123      * Returns a new {@link DefaultChannelPipeline} instance.
 124      */
 125     protected DefaultChannelPipeline newChannelPipeline() {
 126         return new DefaultChannelPipeline(this);
 127     }
 128 
 129     //通道是否可写
 130     @Override
 131     public boolean isWritable() {
 132         //是否存在输出缓冲区且缓冲区可写
 133         ChannelOutboundBuffer buf = unsafe.outboundBuffer();
 134         return buf != null && buf.isWritable();
 135     }
 136 
 137     //获取剩余可写空间
 138     @Override
 139     public long bytesBeforeUnwritable() {
 140         ChannelOutboundBuffer buf = unsafe.outboundBuffer();
 141         // isWritable() is currently assuming if there is no outboundBuffer then the channel is not writable.
 142         // We should be consistent with that here.
 143         return buf != null ? buf.bytesBeforeUnwritable() : 0;
 144     }
 145 
 146     //还需处理多少字节才可写
 147     @Override
 148     public long bytesBeforeWritable() {
 149         ChannelOutboundBuffer buf = unsafe.outboundBuffer();
 150         // isWritable() is currently assuming if there is no outboundBuffer then the channel is not writable.
 151         // We should be consistent with that here.
 152         return buf != null ? buf.bytesBeforeWritable() : Long.MAX_VALUE;
 153     }
 154 
 155     @Override
 156     public Channel parent() {
 157         return parent;
 158     }
 159 
 160     @Override
 161     public ChannelPipeline pipeline() {
 162         return pipeline;
 163     }
 164 
 165     @Override
 166     public ByteBufAllocator alloc() {
 167         return config().getAllocator();
 168     }
 169 
 170     @Override
 171     public EventLoop eventLoop() {
 172         EventLoop eventLoop = this.eventLoop;
 173         if (eventLoop == null) {
 174             throw new IllegalStateException("channel not registered to an event loop");
 175         }
 176         return eventLoop;
 177     }
 178 
 179     @Override
 180     public SocketAddress localAddress() {
 181         SocketAddress localAddress = this.localAddress;
 182         if (localAddress == null) {
 183             try {
 184                 this.localAddress = localAddress = unsafe().localAddress();
 185             } catch (Throwable t) {
 186                 // Sometimes fails on a closed socket in Windows.
 187                 return null;
 188             }
 189         }
 190         return localAddress;
 191     }
 192 
 193     /**
 194      * @deprecated no use-case for this.
 195      */
 196     @Deprecated
 197     protected void invalidateLocalAddress() {
 198         localAddress = null;
 199     }
 200 
 201     //获取远端地址
 202     @Override
 203     public SocketAddress remoteAddress() {
 204         SocketAddress remoteAddress = this.remoteAddress;
 205         if (remoteAddress == null) {
 206             try {
 207                 this.remoteAddress = remoteAddress = unsafe().remoteAddress();
 208             } catch (Throwable t) {
 209                 // Sometimes fails on a closed socket in Windows.
 210                 return null;
 211             }
 212         }
 213         return remoteAddress;
 214     }
 215 
 216     /**
 217      * @deprecated no use-case for this.
 218      */
 219     @Deprecated
 220     protected void invalidateRemoteAddress() {
 221         remoteAddress = null;
 222     }
 223 
 224     //是否已经注册
 225     @Override
 226     public boolean isRegistered() {
 227         return registered;
 228     }
 229 
 230     /******* ChannelOutboundInvoker接口中的方法都交给pipeline对象代理********/
 231     @Override
 232     public ChannelFuture bind(SocketAddress localAddress) {
 233         return pipeline.bind(localAddress);
 234     }
 235 
 236     @Override
 237     public ChannelFuture connect(SocketAddress remoteAddress) {
 238         return pipeline.connect(remoteAddress);
 239     }
 240 
 241     @Override
 242     public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress) {
 243         return pipeline.connect(remoteAddress, localAddress);
 244     }
 245 
 246     @Override
 247     public ChannelFuture disconnect() {
 248         return pipeline.disconnect();
 249     }
 250 
 251     @Override
 252     public ChannelFuture close() {
 253         return pipeline.close();
 254     }
 255 
 256     @Override
 257     public ChannelFuture deregister() {
 258         return pipeline.deregister();
 259     }
 260 
 261     @Override
 262     public Channel flush() {
 263         pipeline.flush();
 264         return this;
 265     }
 266 
 267     @Override
 268     public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
 269         return pipeline.bind(localAddress, promise);
 270     }
 271 
 272     @Override
 273     public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
 274         return pipeline.connect(remoteAddress, promise);
 275     }
 276 
 277     @Override
 278     public ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
 279         return pipeline.connect(remoteAddress, localAddress, promise);
 280     }
 281 
 282     @Override
 283     public ChannelFuture disconnect(ChannelPromise promise) {
 284         return pipeline.disconnect(promise);
 285     }
 286 
 287     @Override
 288     public ChannelFuture close(ChannelPromise promise) {
 289         return pipeline.close(promise);
 290     }
 291 
 292     @Override
 293     public ChannelFuture deregister(ChannelPromise promise) {
 294         return pipeline.deregister(promise);
 295     }
 296 
 297     @Override
 298     public Channel read() {
 299         pipeline.read();
 300         return this;
 301     }
 302 
 303     @Override
 304     public ChannelFuture write(Object msg) {
 305         return pipeline.write(msg);
 306     }
 307 
 308     @Override
 309     public ChannelFuture write(Object msg, ChannelPromise promise) {
 310         return pipeline.write(msg, promise);
 311     }
 312 
 313     @Override
 314     public ChannelFuture writeAndFlush(Object msg) {
 315         return pipeline.writeAndFlush(msg);
 316     }
 317 
 318     @Override
 319     public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
 320         return pipeline.writeAndFlush(msg, promise);
 321     }
 322 
 323     @Override
 324     public ChannelPromise newPromise() {
 325         return pipeline.newPromise();
 326     }
 327 
 328     @Override
 329     public ChannelProgressivePromise newProgressivePromise() {
 330         return pipeline.newProgressivePromise();
 331     }
 332 
 333     @Override
 334     public ChannelFuture newSucceededFuture() {
 335         return pipeline.newSucceededFuture();
 336     }
 337 
 338     @Override
 339     public ChannelFuture newFailedFuture(Throwable cause) {
 340         return pipeline.newFailedFuture(cause);
 341     }
 342 
 343     /**************ChannelOutboundInvoker接口*************/
 344 
 345     //返回ChannelFuture
 346     @Override
 347     public ChannelFuture closeFuture() {
 348         return closeFuture;
 349     }
 350 
 351     //返回UnSafe对象
 352     @Override
 353     public Unsafe unsafe() {
 354         return unsafe;
 355     }
 356 
 357     /**
 358      * Create a new {@link AbstractUnsafe} instance which will be used for the life-time of the {@link Channel}
 359      */
 360     //抽象方法 具体获取Unsafe的方式交给子类实现,
 361     protected abstract AbstractUnsafe newUnsafe();
 362 
 363     /**
 364      * Returns the ID of this channel.
 365      */
 366     @Override
 367     public final int hashCode() {
 368         return id.hashCode();
 369     }
 370 
 371     /**
 372      * Returns {@code true} if and only if the specified object is identical
 373      * with this channel (i.e: {@code this == o}).
 374      */
 375     @Override
 376     public final boolean equals(Object o) {
 377         return this == o;
 378     }
 379 
 380     @Override
 381     public final int compareTo(Channel o) {
 382         if (this == o) {
 383             return 0;
 384         }
 385 
 386         return id().compareTo(o.id());
 387     }
 388 
 389     /**
 390      * Returns the {@link String} representation of this channel.  The returned
 391      * string contains the {@linkplain #hashCode() ID}, {@linkplain #localAddress() local address},
 392      * and {@linkplain #remoteAddress() remote address} of this channel for
 393      * easier identification.
 394      */
 395     @Override
 396     public String toString() {
 397         boolean active = isActive();
 398         if (strValActive == active && strVal != null) {
 399             return strVal;
 400         }
 401 
 402         SocketAddress remoteAddr = remoteAddress();
 403         SocketAddress localAddr = localAddress();
 404         if (remoteAddr != null) {
 405             StringBuilder buf = new StringBuilder(96)
 406                 .append("[id: 0x")
 407                 .append(id.asShortText())
 408                 .append(", L:")
 409                 .append(localAddr)
 410                 .append(active? " - " : " ! ")
 411                 .append("R:")
 412                 .append(remoteAddr)
 413                 .append(‘]‘);
 414             strVal = buf.toString();
 415         } else if (localAddr != null) {
 416             StringBuilder buf = new StringBuilder(64)
 417                 .append("[id: 0x")
 418                 .append(id.asShortText())
 419                 .append(", L:")
 420                 .append(localAddr)
 421                 .append(‘]‘);
 422             strVal = buf.toString();
 423         } else {
 424             StringBuilder buf = new StringBuilder(16)
 425                 .append("[id: 0x")
 426                 .append(id.asShortText())
 427                 .append(‘]‘);
 428             strVal = buf.toString();
 429         }
 430 
 431         strValActive = active;
 432         return strVal;
 433     }
 434 
 435     @Override
 436     public final ChannelPromise voidPromise() {
 437         return pipeline.voidPromise();
 438     }
 439 
 440     /**
 441      * {@link Unsafe} implementation which sub-classes must extend and use.
 442      */
 443     //Unsafe的抽象实现，虽然是抽象类，但没有抽象的方法，只是对某些方法提供了空实现；一些方法的关键实现是交给AbstractChannel的特定子类
 444     protected abstract class AbstractUnsafe implements Unsafe {
 445 
 446         private volatile ChannelOutboundBuffer outboundBuffer = new ChannelOutboundBuffer(AbstractChannel.this);
 447         private RecvByteBufAllocator.Handle recvHandle;
 448         private boolean inFlush0;
 449         /** true if the channel has never been registered, false otherwise */
 450         private boolean neverRegistered = true;
 451 
 452         private void assertEventLoop() {
 453             assert !registered || eventLoop.inEventLoop();
 454         }
 455 
 456         @Override
 457         public RecvByteBufAllocator.Handle recvBufAllocHandle() {
 458             if (recvHandle == null) {
 459                 recvHandle = config().getRecvByteBufAllocator().newHandle();
 460             }
 461             return recvHandle;
 462         }
 463 
 464         @Override
 465         public final ChannelOutboundBuffer outboundBuffer() {
 466             return outboundBuffer;
 467         }
 468 
 469         @Override
 470         public final SocketAddress localAddress() {
 471             return localAddress0();
 472         }
 473 
 474         @Override
 475         public final SocketAddress remoteAddress() {
 476             return remoteAddress0();
 477         }
 478 
 479         //注册；将Unsafe与EventLoop绑定
 480         @Override
 481         public final void register(EventLoop eventLoop, final ChannelPromise promise) {
 482             if (eventLoop == null) {
 483                 throw new NullPointerException("eventLoop");
 484             }
 485             if (isRegistered()) {
 486                 promise.setFailure(new IllegalStateException("registered to an event loop already"));
 487                 return;
 488             }
 489             if (!isCompatible(eventLoop)) {
 490                 promise.setFailure(
 491                         new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
 492                 return;
 493             }
 494 
 495             AbstractChannel.this.eventLoop = eventLoop;
 496 
 497             if (eventLoop.inEventLoop()) {
 498                 //如果当前线程是处理线程，直接注册
 499                 register0(promise);
 500             } else {
 501                 //如果当前线程不是处理线程，则包装成任务丢给eventLoop处理
 502                 try {
 503                     eventLoop.execute(new Runnable() {
 504                         @Override
 505                         public void run() {
 506                             register0(promise);
 507                         }
 508                     });
 509                 } catch (Throwable t) {
 510                     logger.warn(
 511                             "Force-closing a channel whose registration task was not accepted by an event loop: {}",
 512                             AbstractChannel.this, t);
 513                     closeForcibly();
 514                     closeFuture.setClosed();
 515                     safeSetFailure(promise, t);
 516                 }
 517             }
 518         }
 519 
 520         private void register0(ChannelPromise promise) {
 521             try {
 522                 // check if the channel is still open as it could be closed in the mean time when the register
 523                 // call was outside of the eventLoop
 524                 //确认Channel仍然打开
 525                 if (!promise.setUncancellable() || !ensureOpen(promise)) {
 526                     return;
 527                 }
 528 
 529                 boolean firstRegistration = neverRegistered;
 530                 //注册具体流程
 531                 doRegister();
 532                 //修改注册状态
 533                 neverRegistered = false;
 534                 registered = true;
 535 
 536                 // Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
 537                 // user may already fire events through the pipeline in the ChannelFutureListener.
 538                 //产生HandlerAdded
 539                 pipeline.invokeHandlerAddedIfNeeded();
 540 
 541                 safeSetSuccess(promise);
 542                 //注册成功，产生注册事件
 543                 pipeline.fireChannelRegistered();
 544                 
 545                 //激活成功
 546                 if (isActive()) {
 547                     //首次注册，产生active事件
 548                     if (firstRegistration) {
 549                         pipeline.fireChannelActive();
 550                     } else if (config().isAutoRead()) {
 551                         //如果重注册且配置了自动读
 552                         // This channel was registered before and autoRead() is set. This means we need to begin read
 553                         // again so that we process inbound data.
 554                         //
 555                         // See https://github.com/netty/netty/issues/4805
 556                         beginRead();
 557                     }
 558                 }
 559             } catch (Throwable t) {
 560                 // Close the channel directly to avoid FD leak.
 561                 closeForcibly();
 562                 closeFuture.setClosed();
 563                 safeSetFailure(promise, t);
 564             }
 565         }
 566 
 567         //绑定
 568         @Override
 569         public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
 570             assertEventLoop();
 571 
 572             if (!promise.setUncancellable() || !ensureOpen(promise)) {
 573                 return;
 574             }
 575 
 576             // See: https://github.com/netty/netty/issues/576
 577             if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
 578                 localAddress instanceof InetSocketAddress &&
 579                 !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
 580                 !PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) {
 581                 // Warn a user about the fact that a non-root user can‘t receive a
 582                 // broadcast packet on *nix if the socket is bound on non-wildcard address.
 583                 logger.warn(
 584                         "A non-root user can‘t receive a broadcast packet if the socket " +
 585                         "is not bound to a wildcard address; binding to a non-wildcard " +
 586                         "address (" + localAddress + ") anyway as requested.");
 587             }
 588 
 589             boolean wasActive = isActive();
 590             try {
 591                 //具体绑定流程
 592                 doBind(localAddress);
 593             } catch (Throwable t) {
 594                 safeSetFailure(promise, t);
 595                 closeIfClosed();
 596                 return;
 597             }
 598             //绑定成功，产生Active事件
 599             if (!wasActive && isActive()) {
 600                 invokeLater(new Runnable() {
 601                     @Override
 602                     public void run() {
 603                         pipeline.fireChannelActive();
 604                     }
 605                 });
 606             }
 607             //将Promis设置为成功
 608             safeSetSuccess(promise);
 609         }
 610 
 611         //断开连接
 612         @Override
 613         public final void disconnect(final ChannelPromise promise) {
 614             assertEventLoop();
 615 
 616             if (!promise.setUncancellable()) {
 617                 return;
 618             }
 619 
 620             boolean wasActive = isActive();
 621             try {
 622                 //断开具体流程
 623                 doDisconnect();
 624             } catch (Throwable t) {
 625                 safeSetFailure(promise, t);
 626                 closeIfClosed();
 627                 return;
 628             }
 629 
 630             //断开成功，产生Inactive事件
 631             if (wasActive && !isActive()) {
 632                 invokeLater(new Runnable() {
 633                     @Override
 634                     public void run() {
 635                         pipeline.fireChannelInactive();
 636                     }
 637                 });
 638             }
 639 
 640             safeSetSuccess(promise);
 641             closeIfClosed(); // doDisconnect() might have closed the channel
 642         }
 643 
 644         @Override
 645         public final void close(final ChannelPromise promise) {
 646             assertEventLoop();
 647 
 648             close(promise, CLOSE_CLOSED_CHANNEL_EXCEPTION, CLOSE_CLOSED_CHANNEL_EXCEPTION, false);
 649         }
 650 
 651         /**
 652          * Shutdown the output portion of the corresponding {@link Channel}.
 653          * For example this will clean up the {@link ChannelOutboundBuffer} and not allow any more writes.
 654          */
 655         @UnstableApi
 656         public final void shutdownOutput(final ChannelPromise promise) {
 657             assertEventLoop();
 658             shutdownOutput(promise, null);
 659         }
 660 
 661         /**
 662          * Shutdown the output portion of the corresponding {@link Channel}.
 663          * For example this will clean up the {@link ChannelOutboundBuffer} and not allow any more writes.
 664          * @param cause The cause which may provide rational for the shutdown.
 665          */
 666         private void shutdownOutput(final ChannelPromise promise, Throwable cause) {
 667             if (!promise.setUncancellable()) {
 668                 return;
 669             }
 670 
 671             final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
 672             //如果不存在输出缓冲区，设置Promise为失败，抛出已关闭异常
 673             if (  == null) {
 674                 promise.setFailure(CLOSE_CLOSED_CHANNEL_EXCEPTION);
 675                 return;
 676             }
 677             
 678             this.outboundBuffer = null; // Disallow adding any messages and flushes to outboundBuffer.
 679 
 680             //线程关闭原因
 681             final Throwable shutdownCause = cause == null ?
 682                     new ChannelOutputShutdownException("Channel output shutdown") :
 683                     new ChannelOutputShutdownException("Channel output shutdown", cause);
 684             Executor closeExecutor = prepareToClose();
 685             if (closeExecutor != null) {
 686                 closeExecutor.execute(new Runnable() {
 687                     @Override
 688                     public void run() {
 689                         try {
 690                             //关闭流程
 691                             doShutdownOutput();
 692                             promise.setSuccess();
 693                         } catch (Throwable err) {
 694                             promise.setFailure(err);
 695                         } finally {
 696                             // Dispatch to the EventLoop
 697                             eventLoop().execute(new Runnable() {
 698                                 @Override
 699                                 public void run() {
 700                                     closeOutboundBufferForShutdown(pipeline, outboundBuffer, shutdownCause);
 701                                 }
 702                             });
 703                         }
 704                     }
 705                 });
 706             } else {
 707                 try {
 708                     // Execute the shutdown.
 709                     doShutdownOutput();
 710                     promise.setSuccess();
 711                 } catch (Throwable err) {
 712                     promise.setFailure(err);
 713                 } finally {
 714                     closeOutboundBufferForShutdown(pipeline, outboundBuffer, shutdownCause);
 715                 }
 716             }
 717         }
 718 
 719         private void closeOutboundBufferForShutdown(
 720                 ChannelPipeline pipeline, ChannelOutboundBuffer buffer, Throwable cause) {
 721             buffer.failFlushed(cause, false);
 722             buffer.close(cause, true);
 723             pipeline.fireUserEventTriggered(ChannelOutputShutdownEvent.INSTANCE);
 724         }
 725 
 726         private void close(final ChannelPromise promise, final Throwable cause,
 727                            final ClosedChannelException closeCause, final boolean notify) {
 728             if (!promise.setUncancellable()) {
 729                 return;
 730             }
 731 
 732             if (closeInitiated) {
 733                 if (closeFuture.isDone()) {
 734                     // Closed already.
 735                     safeSetSuccess(promise);
 736                 } else if (!(promise instanceof VoidChannelPromise)) { // Only needed if no VoidChannelPromise.
 737                     // This means close() was called before so we just register a listener and return
 738                     closeFuture.addListener(new ChannelFutureListener() {
 739                         @Override
 740                         public void operationComplete(ChannelFuture future) throws Exception {
 741                             promise.setSuccess();
 742                         }
 743                     });
 744                 }
 745                 return;
 746             }
 747 
 748             closeInitiated = true;
 749 
 750             final boolean wasActive = isActive();
 751             final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
 752             this.outboundBuffer = null; // Disallow adding any messages and flushes to outboundBuffer.
 753             Executor closeExecutor = prepareToClose();
 754             if (closeExecutor != null) {
 755                 closeExecutor.execute(new Runnable() {
 756                     @Override
 757                     public void run() {
 758                         try {
 759                             // Execute the close.
 760                             doClose0(promise);
 761                         } finally {
 762                             // Call invokeLater so closeAndDeregister is executed in the EventLoop again!
 763                             invokeLater(new Runnable() {
 764                                 @Override
 765                                 public void run() {
 766                                     if (outboundBuffer != null) {
 767                                         // Fail all the queued messages
 768                                         outboundBuffer.failFlushed(cause, notify);
 769                                         outboundBuffer.close(closeCause);
 770                                     }
 771                                     fireChannelInactiveAndDeregister(wasActive);
 772                                 }
 773                             });
 774                         }
 775                     }
 776                 });
 777             } else {
 778                 try {
 779                     // Close the channel and fail the queued messages in all cases.
 780                     doClose0(promise);
 781                 } finally {
 782                     if (outboundBuffer != null) {
 783                         // Fail all the queued messages.
 784                         outboundBuffer.failFlushed(cause, notify);
 785                         outboundBuffer.close(closeCause);
 786                     }
 787                 }
 788                 if (inFlush0) {
 789                     invokeLater(new Runnable() {
 790                         @Override
 791                         public void run() {
 792                             fireChannelInactiveAndDeregister(wasActive);
 793                         }
 794                     });
 795                 } else {
 796                     fireChannelInactiveAndDeregister(wasActive);
 797                 }
 798             }
 799         }
 800 
 801         private void doClose0(ChannelPromise promise) {
 802             try {
 803                 doClose();
 804                 closeFuture.setClosed();
 805                 safeSetSuccess(promise);
 806             } catch (Throwable t) {
 807                 closeFuture.setClosed();
 808                 safeSetFailure(promise, t);
 809             }
 810         }
 811 
 812         private void fireChannelInactiveAndDeregister(final boolean wasActive) {
 813             deregister(voidPromise(), wasActive && !isActive());
 814         }
 815 
 816         @Override
 817         public final void closeForcibly() {
 818             assertEventLoop();
 819 
 820             try {
 821                 doClose();
 822             } catch (Exception e) {
 823                 logger.warn("Failed to close a channel.", e);
 824             }
 825         }
 826 
 827         @Override
 828         public final void deregister(final ChannelPromise promise) {
 829             assertEventLoop();
 830 
 831             deregister(promise, false);
 832         }
 833 
 834         private void deregister(final ChannelPromise promise, final boolean fireChannelInactive) {
 835             if (!promise.setUncancellable()) {
 836                 return;
 837             }
 838 
 839             if (!registered) {
 840                 safeSetSuccess(promise);
 841                 return;
 842             }
 843 
 844             // As a user may call deregister() from within any method while doing processing in the ChannelPipeline,
 845             // we need to ensure we do the actual deregister operation later. This is needed as for example,
 846             // we may be in the ByteToMessageDecoder.callDecode(...) method and so still try to do processing in
 847             // the old EventLoop while the user already registered the Channel to a new EventLoop. Without delay,
 848             // the deregister operation this could lead to have a handler invoked by different EventLoop and so
 849             // threads.
 850             //
 851             // See:
 852             // https://github.com/netty/netty/issues/4435
 853             invokeLater(new Runnable() {
 854                 @Override
 855                 public void run() {
 856                     try {
 857                         doDeregister();
 858                     } catch (Throwable t) {
 859                         logger.warn("Unexpected exception occurred while deregistering a channel.", t);
 860                     } finally {
 861                         if (fireChannelInactive) {
 862                             pipeline.fireChannelInactive();
 863                         }
 864                         // Some transports like local and AIO does not allow the deregistration of
 865                         // an open channel.  Their doDeregister() calls close(). Consequently,
 866                         // close() calls deregister() again - no need to fire channelUnregistered, so check
 867                         // if it was registered.
 868                         if (registered) {
 869                             registered = false;
 870                             pipeline.fireChannelUnregistered();
 871                         }
 872                         safeSetSuccess(promise);
 873                     }
 874                 }
 875             });
 876         }
 877 
 878         @Override
 879         public final void beginRead() {
 880             assertEventLoop();
 881 
 882             if (!isActive()) {
 883                 return;
 884             }
 885 
 886             try {
 887                 doBeginRead();
 888             } catch (final Exception e) {
 889                 invokeLater(new Runnable() {
 890                     @Override
 891                     public void run() {
 892                         pipeline.fireExceptionCaught(e);
 893                     }
 894                 });
 895                 close(voidPromise());
 896             }
 897         }
 898 
 899         @Override
 900         public final void write(Object msg, ChannelPromise promise) {
 901             assertEventLoop();
 902 
 903             ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
 904             if (outboundBuffer == null) {
 905                 // If the outboundBuffer is null we know the channel was closed and so
 906                 // need to fail the future right away. If it is not null the handling of the rest
 907                 // will be done in flush0()
 908                 // See https://github.com/netty/netty/issues/2362
 909                 safeSetFailure(promise, WRITE_CLOSED_CHANNEL_EXCEPTION);
 910                 // release message now to prevent resource-leak
 911                 ReferenceCountUtil.release(msg);
 912                 return;
 913             }
 914 
 915             int size;
 916             try {
 917                 msg = filterOutboundMessage(msg);
 918                 size = pipeline.estimatorHandle().size(msg);
 919                 if (size < 0) {
 920                     size = 0;
 921                 }
 922             } catch (Throwable t) {
 923                 safeSetFailure(promise, t);
 924                 ReferenceCountUtil.release(msg);
 925                 return;
 926             }
 927 
 928             outboundBuffer.addMessage(msg, size, promise);
 929         }
 930 
 931         @Override
 932         public final void flush() {
 933             assertEventLoop();
 934 
 935             ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
 936             if (outboundBuffer == null) {
 937                 return;
 938             }
 939 
 940             outboundBuffer.addFlush();
 941             flush0();
 942         }
 943 
 944         @SuppressWarnings("deprecation")
 945         protected void flush0() {
 946             if (inFlush0) {
 947                 // Avoid re-entrance
 948                 return;
 949             }
 950 
 951             final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
 952             if (outboundBuffer == null || outboundBuffer.isEmpty()) {
 953                 return;
 954             }
 955 
 956             inFlush0 = true;
 957 
 958             // Mark all pending write requests as failure if the channel is inactive.
 959             if (!isActive()) {
 960                 try {
 961                     if (isOpen()) {
 962                         outboundBuffer.failFlushed(FLUSH0_NOT_YET_CONNECTED_EXCEPTION, true);
 963                     } else {
 964                         // Do not trigger channelWritabilityChanged because the channel is closed already.
 965                         outboundBuffer.failFlushed(FLUSH0_CLOSED_CHANNEL_EXCEPTION, false);
 966                     }
 967                 } finally {
 968                     inFlush0 = false;
 969                 }
 970                 return;
 971             }
 972 
 973             try {
 974                 doWrite(outboundBuffer);
 975             } catch (Throwable t) {
 976                 if (t instanceof IOException && config().isAutoClose()) {
 977                     /**
 978                      * Just call {@link #close(ChannelPromise, Throwable, boolean)} here which will take care of
 979                      * failing all flushed messages and also ensure the actual close of the underlying transport
 980                      * will happen before the promises are notified.
 981                      *
 982                      * This is needed as otherwise {@link #isActive()} , {@link #isOpen()} and {@link #isWritable()}
 983                      * may still return {@code true} even if the channel should be closed as result of the exception.
 984                      */
 985                     close(voidPromise(), t, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false);
 986                 } else {
 987                     try {
 988                         shutdownOutput(voidPromise(), t);
 989                     } catch (Throwable t2) {
 990                         close(voidPromise(), t2, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false);
 991                     }
 992                 }
 993             } finally {
 994                 inFlush0 = false;
 995             }
 996         }
 997 
 998         @Override
 999         public final ChannelPromise voidPromise() {
1000             assertEventLoop();
1001 
1002             return unsafeVoidPromise;
1003         }
1004 
1005         protected final boolean ensureOpen(ChannelPromise promise) {
1006             if (isOpen()) {
1007                 return true;
1008             }
1009 
1010             safeSetFailure(promise, ENSURE_OPEN_CLOSED_CHANNEL_EXCEPTION);
1011             return false;
1012         }
1013 
1014         /**
1015          * Marks the specified {@code promise} as success.  If the {@code promise} is done already, log a message.
1016          */
1017         protected final void safeSetSuccess(ChannelPromise promise) {
1018             if (!(promise instanceof VoidChannelPromise) && !promise.trySuccess()) {
1019                 logger.warn("Failed to mark a promise as success because it is done already: {}", promise);
1020             }
1021         }
1022 
1023         /**
1024          * Marks the specified {@code promise} as failure.  If the {@code promise} is done already, log a message.
1025          */
1026         protected final void safeSetFailure(ChannelPromise promise, Throwable cause) {
1027             if (!(promise instanceof VoidChannelPromise) && !promise.tryFailure(cause)) {
1028                 logger.warn("Failed to mark a promise as failure because it‘s done already: {}", promise, cause);
1029             }
1030         }
1031 
1032         protected final void closeIfClosed() {
1033             if (isOpen()) {
1034                 return;
1035             }
1036             close(voidPromise());
1037         }
1038 
1039         //将任务提交给EventLoop执行
1040         private void invokeLater(Runnable task) {
1041             try {
1042                 // This method is used by outbound operation implementations to trigger an inbound event later.
1043                 // They do not trigger an inbound event immediately because an outbound operation might have been
1044                 // triggered by another inbound event handler method.  If fired immediately, the call stack
1045                 // will look like this for example:
1046                 //
1047                 //   handlerA.inboundBufferUpdated() - (1) an inbound handler method closes a connection.
1048                 //   -> handlerA.ctx.close()
1049                 //      -> channel.unsafe.close()
1050                 //         -> handlerA.channelInactive() - (2) another inbound handler method called while in (1) yet
1051                 //
1052                 // which means the execution of two inbound handler methods of the same handler overlap undesirably.
1053                 eventLoop().execute(task);
1054             } catch (RejectedExecutionException e) {
1055                 logger.warn("Can‘t invoke task later as EventLoop rejected it", e);
1056             }
1057         }
1058 
1059         /**
1060          * Appends the remote address to the message of the exceptions caused by connection attempt failure.
1061          */
1062         protected final Throwable annotateConnectException(Throwable cause, SocketAddress remoteAddress) {
1063             if (cause instanceof ConnectException) {
1064                 return new AnnotatedConnectException((ConnectException) cause, remoteAddress);
1065             }
1066             if (cause instanceof NoRouteToHostException) {
1067                 return new AnnotatedNoRouteToHostException((NoRouteToHostException) cause, remoteAddress);
1068             }
1069             if (cause instanceof SocketException) {
1070                 return new AnnotatedSocketException((SocketException) cause, remoteAddress);
1071             }
1072 
1073             return cause;
1074         }
1075 
1076         /**
1077          * Prepares to close the {@link Channel}. If this method returns an {@link Executor}, the
1078          * caller must call the {@link Executor#execute(Runnable)} method with a task that calls
1079          * {@link #doClose()} on the returned {@link Executor}. If this method returns {@code null},
1080          * {@link #doClose()} must be called from the caller thread. (i.e. {@link EventLoop})
1081          */
1082         protected Executor prepareToClose() {
1083             return null;
1084         }
1085     }
1086 
1087     /**
1088      * Return {@code true} if the given {@link EventLoop} is compatible with this instance.
1089      */
1090     protected abstract boolean isCompatible(EventLoop loop);
1091 
1092     /**
1093      * Returns the {@link SocketAddress} which is bound locally.
1094      */
1095     protected abstract SocketAddress localAddress0();
1096 
1097     /**
1098      * Return the {@link SocketAddress} which the {@link Channel} is connected to.
1099      */
1100     protected abstract SocketAddress remoteAddress0();
1101 
1102     /**
1103      * Is called after the {@link Channel} is registered with its {@link EventLoop} as part of the register process.
1104      *
1105      * Sub-classes may override this method
1106      */
1107     protected void doRegister() throws Exception {
1108         // NOOP
1109     }
1110 
1111     /**
1112      * Bind the {@link Channel} to the {@link SocketAddress}
1113      */
1114     protected abstract void doBind(SocketAddress localAddress) throws Exception;
1115 
1116     /**
1117      * Disconnect this {@link Channel} from its remote peer
1118      */
1119     protected abstract void doDisconnect() throws Exception;
1120 
1121     /**
1122      * Close the {@link Channel}
1123      */
1124     protected abstract void doClose() throws Exception;
1125 
1126     /**
1127      * Called when conditions justify shutting down the output portion of the channel. This may happen if a write
1128      * operation throws an exception.
1129      */
1130     @UnstableApi
1131     protected void doShutdownOutput() throws Exception {
1132         doClose();
1133     }
1134 
1135     /**
1136      * Deregister the {@link Channel} from its {@link EventLoop}.
1137      *
1138      * Sub-classes may override this method
1139      */
1140     protected void doDeregister() throws Exception {
1141         // NOOP
1142     }
1143 
1144     /**
1145      * Schedule a read operation.
1146      */
1147     protected abstract void doBeginRead() throws Exception;
1148 
1149     /**
1150      * Flush the content of the given buffer to the remote peer.
1151      */
1152     protected abstract void doWrite(ChannelOutboundBuffer in) throws Exception;
1153 
1154     /**
1155      * Invoked when a new message is added to a {@link ChannelOutboundBuffer} of this {@link AbstractChannel}, so that
1156      * the {@link Channel} implementation converts the message to another. (e.g. heap buffer -> direct buffer)
1157      */
1158     protected Object filterOutboundMessage(Object msg) throws Exception {
1159         return msg;
1160     }
1161 
1162     static final class CloseFuture extends DefaultChannelPromise {
1163 
1164         CloseFuture(AbstractChannel ch) {
1165             super(ch);
1166         }
1167 
1168         @Override
1169         public ChannelPromise setSuccess() {
1170             throw new IllegalStateException();
1171         }
1172 
1173         @Override
1174         public ChannelPromise setFailure(Throwable cause) {
1175             throw new IllegalStateException();
1176         }
1177 
1178         @Override
1179         public boolean trySuccess() {
1180             throw new IllegalStateException();
1181         }
1182 
1183         @Override
1184         public boolean tryFailure(Throwable cause) {
1185             throw new IllegalStateException();
1186         }
1187 
1188         boolean setClosed() {
1189             return super.trySuccess();
1190         }
1191     }
1192 
1193     private static final class AnnotatedConnectException extends ConnectException {
1194 
1195         private static final long serialVersionUID = 3901958112696433556L;
1196 
1197         AnnotatedConnectException(ConnectException exception, SocketAddress remoteAddress) {
1198             super(exception.getMessage() + ": " + remoteAddress);
1199             initCause(exception);
1200             setStackTrace(exception.getStackTrace());
1201         }
1202 
1203         @Override
1204         public Throwable fillInStackTrace() {
1205             return this;
1206         }
1207     }
1208 
1209     private static final class AnnotatedNoRouteToHostException extends NoRouteToHostException {
1210 
1211         private static final long serialVersionUID = -6801433937592080623L;
1212 
1213         AnnotatedNoRouteToHostException(NoRouteToHostException exception, SocketAddress remoteAddress) {
1214             super(exception.getMessage() + ": " + remoteAddress);
1215             initCause(exception);
1216             setStackTrace(exception.getStackTrace());
1217         }
1218 
1219         @Override
1220         public Throwable fillInStackTrace() {
1221             return this;
1222         }
1223     }
1224 
1225     private static final class AnnotatedSocketException extends SocketException {
1226 
1227         private static final long serialVersionUID = 3896743275010454039L;
1228 
1229         AnnotatedSocketException(SocketException exception, SocketAddress remoteAddress) {
1230             super(exception.getMessage() + ": " + remoteAddress);
1231             initCause(exception);
1232             setStackTrace(exception.getStackTrace());
1233         }
1234 
1235         @Override
1236         public Throwable fillInStackTrace() {
1237             return this;
1238         }
1239     }
1240 }