public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
if (method.getDeclaringClass() == Object.class) {
return method.invoke(invoker, args);
}
/*object#method mock*/
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return invoker.toString();
}
/*...*/
/*省略hashCode和equals代码*/
/*其他实现方法invoke*/
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
复制代码MockClusterInvoker#invoke(Invocation invocation) 装饰者模式，MockClusterInvoker装饰了Invoker，主要看result = this.invoker.invoke(invocation);
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")) {
/*省略代码，log*/
//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 {
/*省略代码，log*/
result = doMockInvoke(invocation, e);
}
}
}
return result;
}
复制代码AbstractClusterInvoker#invoke(final Invocation invocation)，缺省实现FailoverClusterInvoker，首先关注list(invocation);
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
LoadBalance loadbalance = null;
// binding attachments into invocation.
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
((RpcInvocation) invocation).addAttachments(contextAttachments);
}
/*先往下看list方法*/
List<Invoker<T>> invokers = list(invocation);
/*获得负载均衡的具体实现，doInvoke或用到该实例，缺省RandomLoadBalance*/
if (invokers != null && !invokers.isEmpty()) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(RpcUtils.getMethodName(invocation), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
}
/*跳过，默认情况下，将在异步操作中添加调用ID，为了幂等*/
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);
}
复制代码protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
List<Invoker<T>> invokers = directory.list(invocation);
return invokers;
}
复制代码directory.list(invocation);实际调用AbstractDirectory#list(Invocation invocation)，缺省实现为RegistryDirectory
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
if (destroyed) {
throw new RpcException("Directory already destroyed .url: " + getUrl());
}
/*获得invokers，先往下看doList方法*/
List<Invoker<T>> invokers = doList(invocation);
/*获得routers*/
List<Router> localRouters = this.routers; // local reference
if (localRouters != null && !localRouters.isEmpty()) {
/*遍历所有Router，获得正常的invokers*/
for (Router router : localRouters) {
try {
if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, false)) {
invokers = router.route(invokers, getConsumerUrl(), invocation);
}
} catch (Throwable t) {
/*省略代码，log*/
}
}
}
return invokers;
}
复制代码doList(invocation);调用了RegistryDirectory#doList(Invocation invocation)
public List<Invoker<T>> doList(Invocation invocation) {
if (forbidden) {
// 1. No service provider 2. Service providers are disabled
/*省略代码，throw new RpcException*/
}
List<Invoker<T>> invokers = null;
/*缓存了方法和invokers的mapping，Invoker就是具体调用的执行器，以后可以分析怎么获得的*/
Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
String methodName = RpcUtils.getMethodName(invocation);
/*获得入参*/
Object[] args = RpcUtils.getArguments(invocation);
/*获得具体的invokers*/
if (args != null && args.length > 0 && args[0] != null
&& (args[0] instanceof String || args[0].getClass().isEnum())) {
invokers = localMethodInvokerMap.get(methodName + "." + args[0]); // The routing can be enumerated according to the first parameter
}
if (invokers == null) {
invokers = localMethodInvokerMap.get(methodName);
}
if (invokers == null) {
invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
}
if (invokers == null) {
Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
if (iterator.hasNext()) {
invokers = iterator.next();
}
}
}
return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
}
复制代码获得了invokers，我再看上面的AbstractDirectory#list(Invocation invocation)方法，router.route()实际调用了MockInvokersSelector#route(final List<Invoker> invokers, URL url, final Invocation invocation)
public <T> List<Invoker<T>> route(final List<Invoker<T>> invokers,
URL url, final Invocation invocation) throws RpcException {
if (invocation.getAttachments() == null) {
return getNormalInvokers(invokers);
} else {
/*是否需要mock*/
String value = invocation.getAttachments().get(Constants.INVOCATION_NEED_MOCK);
if (value == null)
/*走这*/
return getNormalInvokers(invokers);
else if (Boolean.TRUE.toString().equalsIgnoreCase(value)) {
return getMockedInvokers(invokers);
}
}
return invokers;
}
复制代码private <T> List<Invoker<T>> getNormalInvokers(final List<Invoker<T>> invokers) {
/*如果没有mock的Provider，做校验，校验通过返回所有invokers*/
if (!hasMockProviders(invokers)) {
return invokers;
} else {
/*否则去掉mock的Provider*/
List<Invoker<T>> sInvokers = new ArrayList<Invoker<T>>(invokers.size());
for (Invoker<T> invoker : invokers) {
if (!invoker.getUrl().getProtocol().equals(Constants.MOCK_PROTOCOL)) {
sInvokers.add(invoker);
}
}
return sInvokers;
}
}
复制代码AbstractDirectory#list(Invocation invocation)方法终于结束了，主要就是获得了正常的invokers。 小结：首先Directory获得所有Invokers，然后Router获得所有非mock的Invokers。
接着回到AbstractClusterInvoker#invoke(final Invocation invocation)，获得具体负载均衡的实例后，调用了FailoverClusterInvoker#doInvoke(Invocation invocation, final List<Invoker> invokers, LoadBalance loadbalance)，下面这部分主要为负载到某个Invoker，不想看的可以直接跳到invoke
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
/*检查是否为空*/
checkInvokers(copyinvokers, invocation);
int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
if (len <= 0) {
len = 1;
}
// 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++) {
//在重试之前重新选择以避免Invokers发生变化。
//注意：如果`invokers`改变了，那么`invoked`也会失去准确性。
if (i > 0) {
checkWhetherDestroyed();
copyinvokers = list(invocation);
// check again
checkInvokers(copyinvokers, invocation);
}
/*选择具体某个Invoker，先往下看*/
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
invoked.add(invoker);
RpcContext.getContext().setInvokers((List) invoked);
try {
/*invoke！！！*/
Result result = invoker.invoke(invocation);
if (le != null && logger.isWarnEnabled()) {
/*省略代码，log.warn*/
}
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());
}
}
/*省略代码，throw new RpcException*/
}
复制代码AbstractClusterInvoker#select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected)
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
String methodName = invocation == null ? "" : invocation.getMethodName();
boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
{
//ignore overloaded method
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
stickyInvoker = null;
}
//ignore concurrency problem
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
if (availablecheck && stickyInvoker.isAvailable()) {
return stickyInvoker;
}
}
}
Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);
if (sticky) {
stickyInvoker = invoker;
}
return invoker;
}
复制代码AbstractClusterInvoker#doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected)
private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
/*可有一个直接返回*/
if (invokers.size() == 1)
return invokers.get(0);
if (loadbalance == null) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
/*之前实例化的LoadBalance*/
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
//如果`invoker`在`selected`中或者invoker不可用&& availablecheck为true，则重新选择。
if ((selected != null && selected.contains(invoker))
|| (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
try {
Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
if (rinvoker != null) {
invoker = rinvoker;
} else {
//检查当前所选调用者的索引，如果不是最后一个，选择index+1的这个。
int index = invokers.indexOf(invoker);
try {
//Avoid collision
invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0);
} catch (Exception e) {
/*省略代码，log.warn*/
}
}
} catch (Throwable t) {
/*省略代码，log.error*/
}
}
return invoker;
}
复制代码AbstractLoadBalance#select(List<Invoker<T>> invokers, URL url, Invocation invocation)
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
if (invokers == null || invokers.isEmpty())
return null;
if (invokers.size() == 1)
return invokers.get(0);
return doSelect(invokers, url, invocation);
}
复制代码RandomLoadBalance#doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation)，计算随机值，判断在哪个权重范围内，则返回这个范围中的Invoker
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
int length = invokers.size(); //invokers总数
int totalWeight = 0; //总权重
boolean sameWeight = true; // 权重是都否一样
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
totalWeight += weight; // 累计总权重
if (sameWeight && i > 0
&& weight != getWeight(invokers.get(i - 1), invocation)) {
sameWeight = false; //计算所有权重是否一样
}
}
if (totalWeight > 0 && !sameWeight) {
// 如果（并非每个调用者具有相同的权重并且至少一个调用者的权重>0），则根据totalWeight随机选择。
int offset = random.nextInt(totalWeight);
// 根据随机值返回一个调用者。
for (int i = 0; i < length; i++) {
offset -= getWeight(invokers.get(i), invocation);
if (offset < 0) {
return invokers.get(i);
}
}
}
//如果所有调用者具有相同的权重值或totalWeight = 0，则均匀返回。
return invokers.get(random.nextInt(length));
}
复制代码Invoker获取到了就可以执行了，回到FailoverClusterInvoker#doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance)，接下来invoker.invoke(invocation)，该方法经过了几层装饰，调用责任链（以后再写怎么生成哒），最后调用了AbstractInvoker(实现类DubboInvoker)#invoke(Invocation inv)
public Result invoke(Invocation inv) throws RpcException {
if (destroyed.get()) {
/*省略代码，throw new RpcException*/
}
RpcInvocation invocation = (RpcInvocation) inv;
invocation.setInvoker(this);
if (attachment != null && attachment.size() > 0) {
invocation.addAttachmentsIfAbsent(attachment);
}
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
/**
* invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here,
* because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered
* by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is
* a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information).
*/
invocation.addAttachments(contextAttachments);
}
if (getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY, false)) {
invocation.setAttachment(Constants.ASYNC_KEY, Boolean.TRUE.toString());
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
try {
return doInvoke(invocation);
} catch (InvocationTargetException e) { // biz exception
/*省略，异常处理*/
} catch (RpcException e) {
/*省略，异常处理*/
} catch (Throwable e) {
return new RpcResult(e);
}
}
复制代码DubboInvoker.doInvoke(final Invocation invocation)，发送接收消息
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 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);
RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
return new RpcResult();
} else {
RpcContext.getContext().setFuture(null);
return (Result) currentClient.request(inv, timeout).get();
}
} catch (TimeoutException e) {
/*省略代码，throw new RpcException*/
} catch (RemotingException e) {
/*省略代码，throw new RpcException*/
}
}
复制代码总结：
- 首先通过Directory获得所有Invokers；
- 接着Router获得所有非mock的Invokers；
- 然后通过LoadBalance获得某个Invoker；
- 最后经过一些装饰和责任链后，调用发送接收消息，得到最后结果。
