private void invokeExceptionCaught(final Throwable cause) {
        if (invokeHandler()) {
            try {
//                执行自己实现的异常处理的方法
                handler().exceptionCaught(this, cause);
            } catch (Throwable error) {
                if (logger.isDebugEnabled()) {
                    logger.debug(
                        "An exception {}" +
                        "was thrown by a user handler's exceptionCaught() " +
                        "method while handling the following exception:",
                        ThrowableUtil.stackTraceToString(error), cause);
                } else if (logger.isWarnEnabled()) {
                    logger.warn(
                        "An exception '{}' [enable DEBUG level for full stacktrace] " +
                        "was thrown by a user handler's exceptionCaught() " +
                        "method while handling the following exception:", error, cause);
                }
            }
        } else {
//            出现异常把异常继续向pipeline的下个节点传播
            fireExceptionCaught(cause);
        }
    }
 

/**
 * Create a new {@link ThreadPerChannelEventLoopGroup}.
 *
 * @param maxChannels       the maximum number of channels to handle with this instance. Once you try to register
 *                          a new {@link Channel} and the maximum is exceed it will throw an
 *                          {@link ChannelException} on the {@link #register(Channel)} and
 *                          {@link #register(ChannelPromise)} method.
 *                          Use {@code 0} to use no limit
 * @param executor          the {@link Executor} used to create new {@link Thread} instances that handle the
 *                          registered {@link Channel}s
 * @param args              arguments which will passed to each {@link #newChild(Object...)} call.
 */
protected ThreadPerChannelEventLoopGroup(int maxChannels, Executor executor, Object... args) {
    if (maxChannels < 0) {
        throw new IllegalArgumentException(String.format(
                "maxChannels: %d (expected: >= 0)", maxChannels));
    }
    if (executor == null) {
        throw new NullPointerException("executor");
    }

    if (args == null) {
        childArgs = EmptyArrays.EMPTY_OBJECTS;
    } else {
        childArgs = args.clone();
    }

    this.maxChannels = maxChannels;
    this.executor = executor;

    tooManyChannels = ThrowableUtil.unknownStackTrace(
            new ChannelException("too many channels (max: " + maxChannels + ')'),
            ThreadPerChannelEventLoopGroup.class, "nextChild()");
}
 

private static void loadNativeLibrary() {
    String name = SystemPropertyUtil.get("os.name").toLowerCase(Locale.UK).trim();
    if (!name.startsWith("mac") && !name.contains("bsd") && !name.startsWith("darwin")) {
        throw new IllegalStateException("Only supported on BSD");
    }
    String staticLibName = "netty_transport_native_kqueue";
    String sharedLibName = staticLibName + '_' + PlatformDependent.normalizedArch();
    ClassLoader cl = PlatformDependent.getClassLoader(Native.class);
    try {
        NativeLibraryLoader.load(sharedLibName, cl);
    } catch (UnsatisfiedLinkError e1) {
        try {
            NativeLibraryLoader.load(staticLibName, cl);
            logger.debug("Failed to load {}", sharedLibName, e1);
        } catch (UnsatisfiedLinkError e2) {
            ThrowableUtil.addSuppressed(e1, e2);
            throw e1;
        }
    }
}
 

private static void loadNativeLibrary() {
    String name = SystemPropertyUtil.get("os.name").toLowerCase(Locale.UK).trim();
    if (!name.startsWith("linux")) {
        throw new IllegalStateException("Only supported on Linux");
    }
    String staticLibName = "netty_transport_native_epoll";
    String sharedLibName = staticLibName + '_' + PlatformDependent.normalizedArch();
    ClassLoader cl = PlatformDependent.getClassLoader(Native.class);
    try {
        NativeLibraryLoader.load(sharedLibName, cl);
    } catch (UnsatisfiedLinkError e1) {
        try {
            NativeLibraryLoader.load(staticLibName, cl);
            logger.debug("Failed to load {}", sharedLibName, e1);
        } catch (UnsatisfiedLinkError e2) {
            ThrowableUtil.addSuppressed(e1, e2);
            throw e1;
        }
    }
}
 

@Override
public StreamObserver<Message> invoke(StreamObserver<Message> responseObserver) {
  try {
    this.remote = RpcContext.getContext().getAttachment(Constants.REMOTE_ADDRESS);
    Class<?> requestType = grpcMethodType.requestType();
    PoJo2ProtoStreamObserver servserResponseObserver =
        PoJo2ProtoStreamObserver.newObserverWrap(responseObserver);
    Object result = method.invoke(serviceToInvoke, servserResponseObserver);
    return Proto2PoJoStreamObserver.newObserverWrap((StreamObserver<Object>) result, requestType);
  } catch (Throwable e) {
    String stackTrace = ThrowableUtil.stackTraceToString(e);
    log.error(e.getMessage(), e);
    StatusRuntimeException statusException =
        Status.UNAVAILABLE.withDescription(stackTrace).asRuntimeException();
    responseObserver.onError(statusException);
  } finally {
    log.debug(String.format("Service: %s  Method: %s  RemoteAddress: %s",
        providerUrl.getServiceInterface(), method.getName(), this.remote));
  }
  return null;
}
 

private void streamCall(Message request, StreamObserver<Message> responseObserver) {
  try {
    Class<?> requestType = grpcMethodType.requestType();
    Object reqPojo = SerializerUtil.protobuf2Pojo(request, requestType);
    Object[] requestParams =
        new Object[] {reqPojo, PoJo2ProtoStreamObserver.newObserverWrap(responseObserver)};
    method.invoke(serviceToInvoke, requestParams);
  } catch (Throwable e) {
    String stackTrace = ThrowableUtil.stackTraceToString(e);
    log.error(e.getMessage(), e);
    StatusRuntimeException statusException =
        Status.UNAVAILABLE.withDescription(stackTrace).asRuntimeException();
    responseObserver.onError(statusException);
  } finally {
    log.debug(String.format("Service: %s  Method: %s  RemoteAddress: %s",
        providerUrl.getServiceInterface(), method.getName(), this.remote));
  }
}
 

@Override
public void onNext(Object value) {
  try {
    Object respPojo = value;
    Message respProtoBufer = SerializerUtil.pojo2Protobuf(respPojo);
    streamObserver.onNext(respProtoBufer);
  } catch (ProtobufException e) {
    String stackTrace = ThrowableUtil.stackTraceToString(e);
    StatusRuntimeException statusException =
        Status.UNAVAILABLE.withDescription(stackTrace).asRuntimeException();
    streamObserver.onError(statusException);
  }
}
 

@Override
public void onNext(Message value) {
  try {
    Object respPoJo = SerializerUtil.protobuf2Pojo(value, poJoType);
    streamObserver.onNext(respPoJo);
  } catch (ProtobufException e) {
    String stackTrace = ThrowableUtil.stackTraceToString(e);
    StatusRuntimeException statusException =
        Status.UNAVAILABLE.withDescription(stackTrace).asRuntimeException();
    streamObserver.onError(statusException);
  }
}
 

@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    if (inboundHandler == null) {
        throw new IllegalStateException(
                "init() must be invoked before being added to a " + ChannelPipeline.class.getSimpleName() +
                        " if " +  CombinedChannelDuplexHandler.class.getSimpleName() +
                        " was constructed with the default constructor.");
    }

    outboundCtx = new DelegatingChannelHandlerContext(ctx, outboundHandler);
    inboundCtx = new DelegatingChannelHandlerContext(ctx, inboundHandler) {
        @SuppressWarnings("deprecation")
        @Override
        public ChannelHandlerContext fireExceptionCaught(Throwable cause) {
            if (!outboundCtx.removed) {
                try {
                    // We directly delegate to the ChannelOutboundHandler as this may override exceptionCaught(...)
                    // as well
                    outboundHandler.exceptionCaught(outboundCtx, cause);
                } catch (Throwable error) {
                    if (logger.isDebugEnabled()) {
                        logger.debug(
                                "An exception {}" +
                                "was thrown by a user handler's exceptionCaught() " +
                                "method while handling the following exception:",
                                ThrowableUtil.stackTraceToString(error), cause);
                    } else if (logger.isWarnEnabled()) {
                        logger.warn(
                                "An exception '{}' [enable DEBUG level for full stacktrace] " +
                                "was thrown by a user handler's exceptionCaught() " +
                                "method while handling the following exception:", error, cause);
                    }
                }
            } else {
                super.fireExceptionCaught(cause);
            }
            return this;
        }
    };

    // The inboundCtx and outboundCtx were created and set now it's safe to call removeInboundHandler() and
    // removeOutboundHandler().
    handlerAdded = true;

    try {
        inboundHandler.handlerAdded(inboundCtx);
    } finally {
        outboundHandler.handlerAdded(outboundCtx);
    }
}