protected synchronized void doExport() {
	if (unexported) {
		throw new IllegalStateException("Already unexported!");
	}
	if (exported) {
		return;
	}
	exported = true;
	if (interfaceName == null || interfaceName.length() == 0) {
		throw new IllegalStateException("<dubbo:service interface=\"\" /> interface not allow null!");
	}
	checkDefault();
	if (provider != null) {
		if (application == null) {
			application = provider.getApplication();
		}
		if (module == null) {
			module = provider.getModule();
		}
		if (registries == null) {
			registries = provider.getRegistries();
		}
		if (monitor == null) {
			monitor = provider.getMonitor();
		}
		if (protocols == null) {
			protocols = provider.getProtocols();
		}
	}
	if (module != null) {
		if (registries == null) {
			registries = module.getRegistries();
		}
		if (monitor == null) {
			monitor = module.getMonitor();
		}
	}
	if (application != null) {
		if (registries == null) {
			registries = application.getRegistries();
		}
		if (monitor == null) {
			monitor = application.getMonitor();
		}
	}
	if (ref instanceof GenericService) {
		interfaceClass = GenericService.class;
		if (StringUtils.isEmpty(generic)) {
			generic = Boolean.TRUE.toString();
		}
	} else {
		try {
			interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
					.getContextClassLoader());
		} catch (ClassNotFoundException e) {
			throw new IllegalStateException(e.getMessage(), e);
		}
		checkInterfaceAndMethods(interfaceClass, methods);
		checkRef();
		generic = Boolean.FALSE.toString();
	}
	if (local != null) {
		if ("true".equals(local)) {
			local = interfaceName + "Local";
		}
		Class<?> localClass;
		try {
			localClass = ClassHelper.forNameWithThreadContextClassLoader(local);
		} catch (ClassNotFoundException e) {
			throw new IllegalStateException(e.getMessage(), e);
		}
		if (!interfaceClass.isAssignableFrom(localClass)) {
			throw new IllegalStateException("The local implementation class " + localClass.getName() + " not implement interface " + interfaceName);
		}
	}
	if (stub != null) {
		if ("true".equals(stub)) {
			stub = interfaceName + "Stub";
		}
		Class<?> stubClass;
		try {
			stubClass = ClassHelper.forNameWithThreadContextClassLoader(stub);
		} catch (ClassNotFoundException e) {
			throw new IllegalStateException(e.getMessage(), e);
		}
		if (!interfaceClass.isAssignableFrom(stubClass)) {
			throw new IllegalStateException("The stub implementation class " + stubClass.getName() + " not implement interface " + interfaceName);
		}
	}
	checkApplication();
	checkRegistry();
	checkProtocol();
	appendProperties(this);
	checkStubAndMock(interfaceClass);
	if (path == null || path.length() == 0) {
		path = interfaceName;
	}
	doExportUrls();
	ProviderModel providerModel = new ProviderModel(getUniqueServiceName(), this, ref);
	ApplicationModel.initProviderModel(getUniqueServiceName(), providerModel);
}

private void doExportUrls() {
        List<URL> registryURLs = loadRegistries(true);
        for (ProtocolConfig protocolConfig : protocols) {
            doExportUrlsFor1Protocol(protocolConfig, registryURLs);
        }
    }

String host = this.findConfigedHosts(protocolConfig, registryURLs, map);
Integer port = this.findConfigedPorts(protocolConfig, name, map);
URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);

if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
		.hasExtension(url.getProtocol())) {
	url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
			.getExtension(url.getProtocol()).getConfigurator(url).configure(url);
}

String scope = url.getParameter(Constants.SCOPE_KEY);
// don't export when none is configured
if (!Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {

	// export to local if the config is not remote (export to remote only when config is remote)
	if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
		exportLocal(url);
	}
	// export to remote if the config is not local (export to local only when config is local)
	if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) {
		if (logger.isInfoEnabled()) {
			logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
		}
		if (registryURLs != null && registryURLs.size() > 0) {
			for (URL registryURL : registryURLs) {
				url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
				URL monitorUrl = loadMonitor(registryURL);
				if (monitorUrl != null) {
					url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
				}
				if (logger.isInfoEnabled()) {
					logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
				}
				Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
				DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

				Exporter<?> exporter = protocol.export(wrapperInvoker);
				exporters.add(exporter);
			}
		} else {
			Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
			DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);

			Exporter<?> exporter = protocol.export(wrapperInvoker);
			exporters.add(exporter);
		}
	}
}
this.urls.add(url);

if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
	URL local = URL.valueOf(url.toFullString())
			.setProtocol(Constants.LOCAL_PROTOCOL)
			.setHost(LOCALHOST)
			.setPort(0);
	ServiceClassHolder.getInstance().pushServiceClass(getServiceClass(ref));
	Exporter<?> exporter = protocol.export(
			proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
	exporters.add(exporter);
	logger.info("Export dubbo service " + interfaceClass.getName() + " to local registry");
}

// JavassistProxyFactory
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
        // TODO Wrapper cannot handle this scenario correctly: the classname contains '$'
        final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
        return new AbstractProxyInvoker<T>(proxy, type, url) {
            @Override
            protected Object doInvoke(T proxy, String methodName,
                                      Class<?>[] parameterTypes,
                                      Object[] arguments) throws Throwable {
                return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
            }
        };
    }

// ProtocolListenerWrapper 返回ListenerExporterWrapper(Exporter<T>, List<ExporterListener>)
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
	if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
		return protocol.export(invoker);
	}
	return new ListenerExporterWrapper<T>(protocol.export(invoker),
			Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class)
					.getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY)));
}

// ProtocolFilterWrappe
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
	if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
		return protocol.export(invoker);
	}
	return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));
}

// InjvmProtocol 返回InjvmExporter(Invoker<T>, String, Map<String, Exporter<?>>)
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
	return new InjvmExporter<T>(invoker, invoker.getUrl().getServiceKey(), exporterMap);
}

// DubboProtocol
createServer(URL url) ->
server = Exchangers.bind(url, requestHandler)

// Exchangers
return getExchanger(url).bind(url, handler);

// HeaderExchanger
return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
// startHeatbeatTimer()

// Transporters
// 有Grizzly、Mina、Netty4、Netty(走了这个)
return getTransporter().bind(url, handler);

// NettyTransporter
return new NettyServer(url, listener);

// 创建server监听，NettyHandler处理Channel的连接/关闭，读/写事件
NettyServer.doOpen()
// pipeline.addLast("handler", nettyHandler)

// RegistryProtocol
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
	//export invoker
	final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);

	URL registryUrl = getRegistryUrl(originInvoker);

	//registry provider
	final Registry registry = getRegistry(originInvoker);
	final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);

	//to judge to delay publish whether or not
	boolean register = registedProviderUrl.getParameter("register", true);

	ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registedProviderUrl);

	if (register) {
		register(registryUrl, registedProviderUrl);
		ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
	}

	// Subscribe the override data
	// FIXME When the provider subscribes, it will affect the scene : a certain JVM exposes the service and call the same service. Because the subscribed is cached key with the name of the service, it causes the subscription information to cover.
	// 当提供者订阅时，它将影响场景：某个JVM公开服务并调用相同的服务。因为订阅是用服务的名称缓存的，所以它会导致订阅信息被覆盖。
	final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
	final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
	overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
	registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
	//Ensure that a new exporter instance is returned every time export
	return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registedProviderUrl);
}

List<URL> urls = new ArrayList<URL>();
for (String path : toCategoriesPath(url)) {
	ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
	if (listeners == null) {
		zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
		listeners = zkListeners.get(url);
	}
	ChildListener zkListener = listeners.get(listener);
	if (zkListener == null) {
		listeners.putIfAbsent(listener, new ChildListener() {
			public void childChanged(String parentPath, List<String> currentChilds) {
				ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
			}
		});
		zkListener = listeners.get(listener);
	}
	zkClient.create(path, false);
	List<String> children = zkClient.addChildListener(path, zkListener);
	if (children != null) {
		urls.addAll(toUrlsWithEmpty(url, path, children));
	}
}
notify(url, listener, urls);

private T createProxy(Map<String, String> map) {
	URL tmpUrl = new URL("temp", "localhost", 0, map);
	final boolean isJvmRefer;
	if (isInjvm() == null) {
		if (url != null && url.length() > 0) { // if a url is specified, don't do local reference
			isJvmRefer = false;
		} else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
			// by default, reference local service if there is
			isJvmRefer = true;
		} else {
			isJvmRefer = false;
		}
	} else {
		isJvmRefer = isInjvm().booleanValue();
	}

	if (isJvmRefer) {
		URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
		invoker = refprotocol.refer(interfaceClass, url);
		if (logger.isInfoEnabled()) {
			logger.info("Using injvm service " + interfaceClass.getName());
		}
	} else {
		if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
			String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
			if (us != null && us.length > 0) {
				for (String u : us) {
					URL url = URL.valueOf(u);
					if (url.getPath() == null || url.getPath().length() == 0) {
						url = url.setPath(interfaceName);
					}
					if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
						urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
					} else {
						urls.add(ClusterUtils.mergeUrl(url, map));
					}
				}
			}
		} else { // assemble URL from register center's configuration
			List<URL> us = loadRegistries(false);
			if (us != null && !us.isEmpty()) {
				for (URL u : us) {
					URL monitorUrl = loadMonitor(u);
					if (monitorUrl != null) {
						map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
					}
					urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
				}
			}
			if (urls.isEmpty()) {
				throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address=\"...\" /> to your spring config.");
			}
		}

		if (urls.size() == 1) {
			invoker = refprotocol.refer(interfaceClass, urls.get(0));
		} else {
			List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
			URL registryURL = null;
			for (URL url : urls) {
				invokers.add(refprotocol.refer(interfaceClass, url));
				if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
					registryURL = url; // use last registry url
				}
			}
			if (registryURL != null) { // registry url is available
				// use AvailableCluster only when register's cluster is available
				URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
				invoker = cluster.join(new StaticDirectory(u, invokers));
			} else { // not a registry url
				invoker = cluster.join(new StaticDirectory(invokers));
			}
		}
	}

	Boolean c = check;
	if (c == null && consumer != null) {
		c = consumer.isCheck();
	}
	if (c == null) {
		c = true; // default true
	}
	if (c && !invoker.isAvailable()) {
		throw new IllegalStateException("Failed to check the status of the service " + interfaceName + ". No provider available for the service " + (group == null ? "" : group + "/") + interfaceName + (version == null ? "" : ":" + version) + " from the url " + invoker.getUrl() + " to the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
	}
	if (logger.isInfoEnabled()) {
		logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
	}
	// create service proxy
	return (T) proxyFactory.getProxy(invoker);
}

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
	url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
	Registry registry = registryFactory.getRegistry(url);
	if (RegistryService.class.equals(type)) {
		return proxyFactory.getInvoker((T) registry, type, url);
	}

	// group="a,b" or group="*"
	Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
	String group = qs.get(Constants.GROUP_KEY);
	if (group != null && group.length() > 0) {
		if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1
				|| "*".equals(group)) {
			return doRefer(getMergeableCluster(), registry, type, url);
		}
	}
	return doRefer(cluster, registry, type, url);
}

// super(url) -> FailbackRegistry
doRegister(url);
// -> ZookeeperRegistry
zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));

// ZookeeperRegistry构造方法
super(url);
if (url.isAnyHost()) {
	throw new IllegalStateException("registry address == null");
}
String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
if (!group.startsWith(Constants.PATH_SEPARATOR)) {
	group = Constants.PATH_SEPARATOR + group;
}
this.root = group;
// 创建连接
zkClient = zookeeperTransporter.connect(url);
// 添加监听器，状态改变时恢复register与subscribe
zkClient.addStateListener(new StateListener() {
	@Override
	public void stateChanged(int state) {
		if (state == RECONNECTED) {
			try {
				recover();
			} catch (Exception e) {
				logger.error(e.getMessage(), e);
			}
		}
	}
});

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
	RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
	directory.setRegistry(registry);
	directory.setProtocol(protocol);
	// all attributes of REFER_KEY
	Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
	URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters);
	if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
			&& url.getParameter(Constants.REGISTER_KEY, true)) {
		registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
				Constants.CHECK_KEY, String.valueOf(false)));
	}
	directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
			Constants.PROVIDERS_CATEGORY
					+ "," + Constants.CONFIGURATORS_CATEGORY
					+ "," + Constants.ROUTERS_CATEGORY));

	Invoker invoker = cluster.join(directory);
	ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
	return invoker;
}

public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
	optimizeSerialization(url);
	// create rpc invoker.
	DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
	invokers.add(invoker);
	return invoker;
}

public Result invoke(Invocation invocation) throws RpcException {
	Result result = null;

	String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
	if (value.length() == 0 || value.equalsIgnoreCase("false")) {
		//no mock
		result = this.invoker.invoke(invocation);
	} else if (value.startsWith("force")) {
		if (logger.isWarnEnabled()) {
			logger.info("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + directory.getUrl());
		}
		//force:direct mock
		result = doMockInvoke(invocation, null);
	} else {
		//fail-mock
		try {
			result = this.invoker.invoke(invocation);
		} catch (RpcException e) {
			if (e.isBiz()) {
				throw e;
			} else {
				if (logger.isWarnEnabled()) {
					logger.warn("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getUrl(), e);
				}
				result = doMockInvoke(invocation, e);
			}
		}
	}
	return result;
}

// retry loop.
RpcException le = null; // last exception.
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers.
Set<String> providers = new HashSet<String>(len);
for (int i = 0; i < len; i++) {
	//Reselect before retry to avoid a change of candidate `invokers`.
	//NOTE: if `invokers` changed, then `invoked` also lose accuracy.
	if (i > 0) {
		checkWhetherDestroyed();
		copyinvokers = list(invocation);
		// check again
		checkInvokers(copyinvokers, invocation);
	}
	Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
	invoked.add(invoker);
	RpcContext.getContext().setInvokers((List) invoked);
	try {
		Result result = invoker.invoke(invocation);
		if (le != null && logger.isWarnEnabled()) {
			logger.warn("Although retry the method " + invocation.getMethodName()
					+ " in the service " + getInterface().getName()
					+ " was successful by the provider " + invoker.getUrl().getAddress()
					+ ", but there have been failed providers " + providers
					+ " (" + providers.size() + "/" + copyinvokers.size()
					+ ") from the registry " + directory.getUrl().getAddress()
					+ " on the consumer " + NetUtils.getLocalHost()
					+ " using the dubbo version " + Version.getVersion() + ". Last error is: "
					+ le.getMessage(), le);
		}
		return result;
	} catch (RpcException e) {
		if (e.isBiz()) { // biz exception.
			throw e;
		}
		le = e;
	} catch (Throwable e) {
		le = new RpcException(e.getMessage(), e);
	} finally {
		providers.add(invoker.getUrl().getAddress());
	}
}

protected Result doInvoke(final Invocation invocation) throws Throwable {
	RpcInvocation inv = (RpcInvocation) invocation;
	final String methodName = RpcUtils.getMethodName(invocation);
	inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
	inv.setAttachment(Constants.VERSION_KEY, version);

	ExchangeClient currentClient;
	if (clients.length == 1) {
		currentClient = clients[0];
	} else {
		currentClient = clients[index.getAndIncrement() % clients.length];
	}
	try {
		boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
		boolean isAsyncFuture = RpcUtils.isGeneratedFuture(inv) || RpcUtils.isFutureReturnType(inv);
		boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
		int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
		if (isOneway) {
			boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
			currentClient.send(inv, isSent);
			RpcContext.getContext().setFuture(null);
			return new RpcResult();
		} else if (isAsync) {
			ResponseFuture future = currentClient.request(inv, timeout);
			// For compatibility
			FutureAdapter<Object> futureAdapter = new FutureAdapter<>(future);
			RpcContext.getContext().setFuture(futureAdapter);

			Result result;
			if (isAsyncFuture) {
				// register resultCallback, sometimes we need the asyn result being processed by the filter chain.
				result = new AsyncRpcResult(futureAdapter, futureAdapter.getResultFuture(), false);
			} else {
				result = new SimpleAsyncRpcResult(futureAdapter, futureAdapter.getResultFuture(), false);
			}
			return result;
		} else {
			RpcContext.getContext().setFuture(null);
			return (Result) currentClient.request(inv, timeout).get();
		}
	} catch (TimeoutException e) {
		throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
	} catch (RemotingException e) {
		throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
	}
}

// NettyCodecAdapter
protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
	org.apache.dubbo.remoting.buffer.ChannelBuffer buffer = new NettyBackedChannelBuffer(out);
	Channel ch = ctx.channel();
	NettyChannel channel = NettyChannel.getOrAddChannel(ch, url, handler);
	try {
		codec.encode(channel, buffer, msg);
	} finally {
		NettyChannel.removeChannelIfDisconnected(ch);
	}
}


// ExchangeCodec
protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
	Serialization serialization = getSerialization(channel);
	// header.
	byte[] header = new byte[HEADER_LENGTH];
	// set magic number.
	Bytes.short2bytes(MAGIC, header);

	// set request and serialization flag.
	header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());

	if (req.isTwoWay()) header[2] |= FLAG_TWOWAY;
	if (req.isEvent()) header[2] |= FLAG_EVENT;

	// set request id.
	Bytes.long2bytes(req.getId(), header, 4);

	// encode request data.
	int savedWriteIndex = buffer.writerIndex();
	buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
	ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
	ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
	if (req.isEvent()) {
		encodeEventData(channel, out, req.getData());
	} else {
		encodeRequestData(channel, out, req.getData(), req.getVersion());
	}
	out.flushBuffer();
	if (out instanceof Cleanable) {
		((Cleanable) out).cleanup();
	}
	bos.flush();
	bos.close();
	int len = bos.writtenBytes();
	checkPayload(channel, len);
	Bytes.int2bytes(len, header, 12);

	// write
	buffer.writerIndex(savedWriteIndex);
	buffer.writeBytes(header); // write header.
	buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
}

protected void decode(ChannelHandlerContext ctx, ByteBuf input, List<Object> out) throws Exception {
	ChannelBuffer message = new NettyBackedChannelBuffer(input);
	NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);
	Object msg;
	int saveReaderIndex;
	try {
		// decode object.
		do {
			saveReaderIndex = message.readerIndex();
			try {
				msg = codec.decode(channel, message);
			} catch (IOException e) {
				throw e;
			}
			if (msg == Codec2.DecodeResult.NEED_MORE_INPUT) {
				message.readerIndex(saveReaderIndex);
				break;
			} else {
				//is it possible to go here ?
				if (saveReaderIndex == message.readerIndex()) {
					throw new IOException("Decode without read data.");
				}
				if (msg != null) {
					out.add(msg);
				}
			}
		} while (message.readable());
	} finally {
		NettyChannel.removeChannelIfDisconnected(ctx.channel());
	}
}

public Object decode(Channel channel, InputStream input) throws IOException {
	ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
			.deserialize(channel.getUrl(), input);

	String dubboVersion = in.readUTF();
	request.setVersion(dubboVersion);
	setAttachment(Constants.DUBBO_VERSION_KEY, dubboVersion);

	setAttachment(Constants.PATH_KEY, in.readUTF());
	setAttachment(Constants.VERSION_KEY, in.readUTF());

	setMethodName(in.readUTF());
	try {
		Object[] args;
		Class<?>[] pts;
		String desc = in.readUTF();
		if (desc.length() == 0) {
			pts = DubboCodec.EMPTY_CLASS_ARRAY;
			args = DubboCodec.EMPTY_OBJECT_ARRAY;
		} else {
			pts = ReflectUtils.desc2classArray(desc);
			args = new Object[pts.length];
			for (int i = 0; i < args.length; i++) {
				try {
					args[i] = in.readObject(pts[i]);
				} catch (Exception e) {
					if (log.isWarnEnabled()) {
						log.warn("Decode argument failed: " + e.getMessage(), e);
					}
				}
			}
		}
		setParameterTypes(pts);

		Map<String, String> map = (Map<String, String>) in.readObject(Map.class);
		if (map != null && map.size() > 0) {
			Map<String, String> attachment = getAttachments();
			if (attachment == null) {
				attachment = new HashMap<String, String>();
			}
			attachment.putAll(map);
			setAttachments(attachment);
		}
		//decode argument ,may be callback
		for (int i = 0; i < args.length; i++) {
			args[i] = decodeInvocationArgument(channel, this, pts, i, args[i]);
		}

		setArguments(args);

	} catch (ClassNotFoundException e) {
		throw new IOException(StringUtils.toString("Read invocation data failed.", e));
	} finally {
		if (in instanceof Cleanable) {
			((Cleanable) in).cleanup();
		}
	}
	return this;
}

public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
	int save = buffer.readerIndex();
	MultiMessage result = MultiMessage.create();
	do {
		Object obj = codec.decode(channel, buffer);
		if (Codec2.DecodeResult.NEED_MORE_INPUT == obj) {
			buffer.readerIndex(save);
			break;
		} else {
			result.addMessage(obj);
			logMessageLength(obj, buffer.readerIndex() - save);
			save = buffer.readerIndex();
		}
	} while (true);
	if (result.isEmpty()) {
		return Codec2.DecodeResult.NEED_MORE_INPUT;
	}
	if (result.size() == 1) {
		return result.get(0);
	}
	return result;
}

public CompletableFuture<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
	if (message instanceof Invocation) {
		Invocation inv = (Invocation) message;
		Invoker<?> invoker = getInvoker(channel, inv);
		// need to consider backward-compatibility if it's a callback
		if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
			String methodsStr = invoker.getUrl().getParameters().get("methods");
			boolean hasMethod = false;
			if (methodsStr == null || !methodsStr.contains(",")) {
				hasMethod = inv.getMethodName().equals(methodsStr);
			} else {
				String[] methods = methodsStr.split(",");
				for (String method : methods) {
					if (inv.getMethodName().equals(method)) {
						hasMethod = true;
						break;
					}
				}
			}
			if (!hasMethod) {
				logger.warn(new IllegalStateException("The methodName " + inv.getMethodName()
						+ " not found in callback service interface ,invoke will be ignored."
						+ " please update the api interface. url is:"
						+ invoker.getUrl()) + " ,invocation is :" + inv);
				return null;
			}
		}
		RpcContext rpcContext = RpcContext.getContext();
		boolean supportServerAsync = invoker.getUrl().getMethodParameter(inv.getMethodName(), Constants.ASYNC_KEY, false);
		if (supportServerAsync) {
			CompletableFuture<Object> future = new CompletableFuture<>();
			rpcContext.setAsyncContext(new AsyncContextImpl(future));
		}
		rpcContext.setRemoteAddress(channel.getRemoteAddress());
		Result result = invoker.invoke(inv);

		if (result instanceof AsyncRpcResult) {
			return ((AsyncRpcResult) result).getResultFuture().thenApply(r -> (Object) r);
		} else {
			return CompletableFuture.completedFuture(result);
		}
	}
	throw new RemotingException(channel, "Unsupported request: "
			+ (message == null ? null : (message.getClass().getName() + ": " + message))
			+ ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
}

/** 
按照顺序decoder先执行对服务端传过来的数据进行解码，解析出序列化协议、响应状态、响应id(即请求id)。
把响应body数据读到DecodeableRpcResult对象中，进行解析同时加载处理原始Request数据，
这个Request对象在请求时会被缓存到DefaultFuture中，加载Request的目的是因为Request中Invocation中携带了服务接口的返回值类型信息，
需要根据这个类型把响应解析创建对应类型的对象。此步在DecodeableRpcResult.decode中实现。
*/
protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
	byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
	Serialization s = CodecSupport.getSerialization(channel.getUrl(), proto);
	// get request id.
	long id = Bytes.bytes2long(header, 4);
	if ((flag & FLAG_REQUEST) == 0) {
		// decode response.
		Response res = new Response(id);
		if ((flag & FLAG_EVENT) != 0) {
			res.setEvent(Response.HEARTBEAT_EVENT);
		}
		// get status.
		byte status = header[3];
		res.setStatus(status);
		if (status == Response.OK) {
			try {
				Object data;
				if (res.isHeartbeat()) {
					data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is));
				} else if (res.isEvent()) {
					data = decodeEventData(channel, deserialize(s, channel.getUrl(), is));
				} else {
					DecodeableRpcResult result;
					if (channel.getUrl().getParameter(
							Constants.DECODE_IN_IO_THREAD_KEY,
							Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
						result = new DecodeableRpcResult(channel, res, is,
								(Invocation) getRequestData(id), proto);
						// 执行DecodeableRpcResult.decode
						result.decode();
					} else {
						result = new DecodeableRpcResult(channel, res,
								new UnsafeByteArrayInputStream(readMessageData(is)),
								(Invocation) getRequestData(id), proto);
					}
					data = result;
				}
				res.setResult(data);
			} catch (Throwable t) {
				if (log.isWarnEnabled()) {
					log.warn("Decode response failed: " + t.getMessage(), t);
				}
				res.setStatus(Response.CLIENT_ERROR);
				res.setErrorMessage(StringUtils.toString(t));
			}
		} else {
			res.setErrorMessage(deserialize(s, channel.getUrl(), is).readUTF());
		}
		return res;
	} else {
		// decode request.
		Request req = new Request(id);
		req.setVersion(Version.getProtocolVersion());
		req.setTwoWay((flag & FLAG_TWOWAY) != 0);
		if ((flag & FLAG_EVENT) != 0) {
			req.setEvent(Request.HEARTBEAT_EVENT);
		}
		try {
			Object data;
			if (req.isHeartbeat()) {
				data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is));
			} else if (req.isEvent()) {
				data = decodeEventData(channel, deserialize(s, channel.getUrl(), is));
			} else {
				DecodeableRpcInvocation inv;
				if (channel.getUrl().getParameter(
						Constants.DECODE_IN_IO_THREAD_KEY,
						Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
					inv = new DecodeableRpcInvocation(channel, req, is, proto);
					inv.decode();
				} else {
					inv = new DecodeableRpcInvocation(channel, req,
							new UnsafeByteArrayInputStream(readMessageData(is)), proto);
				}
				data = inv;
			}
			req.setData(data);
		} catch (Throwable t) {
			if (log.isWarnEnabled()) {
				log.warn("Decode request failed: " + t.getMessage(), t);
			}
			// bad request
			req.setBroken(true);
			req.setData(t);
		}
		return req;
	}
}