由Dubbo路由引发的一言不合就看源码
Posted 百里马
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了由Dubbo路由引发的一言不合就看源码相关的知识，希望对你有一定的参考价值。
由Router引入
由于业务上的需要，很早之前就想自己扩展Dubbo的路由功能。But，之前试了一下，发现好像并没有找到正确的打开方式，而且由于时间不是很充足，加上自己心浮气躁浅尝辄止，于是乎就放弃了这块的研究。
不过今日，在手头上一块比较忙的事情告一段落了之后，又开始漫漫的research之路。。
好了，下面先来说说之前被困在了什么地方。
首先吐槽一下，官方文档对于这一块的用法写得非常的不全面，只介绍了条件路由的语法，而没有告诉我们怎么样使其生效。当然最简单的一种做法，官方文档上也说了，装个dubbo-admin在其中设置就好了。但是如果是想自定义扩展的话呢？这个时候我们来看看Developer Guide是怎么写的：
没错，就是这里的 (3)扩展配置 误导了我，导致了之前的放弃。因为这里的扩展配置，完全就是错的。今天看了源码之后，发现完全就没有这么一个配置项，关于正确的路由规则的设置方法下面会讲。
好了，继续看，那我们就先看看条件路由的语法，然而，又发现一个官方文档的bug：
官方文档写的是runtime不填缺省为false，而Dubbo的代码中呢？
有点儿醉！接着看吧；既然看了代码，我们就从整体上来看看Dubbo的结构，这里从Consumer入手，来看看它是怎么找到合适的Provider来调用的。
源码分析
注意，下面我们会摒弃Mock、Generic等非主流程，精简代码来分析。
核心类ReferenceConfig
好啦，从Consumer入手，我们看看到底发生了什么，首先，Dubbo会代理我们在Consumer中引用的接口，一般都用<dubbo:reference />来配置，具体的代理逻辑在ReferenceConfig中：
private T createProxy(Map<String, String> map) {
// 通过注册中心配置拼装URL，这里省略了inJvm、点对点直连等
List<URL> us = loadRegistries(false);
if (us != null && us.size() > 0) {
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)));
}
}
// 这里的url是注册中心的url
if (urls == null || urls.size() == 0) {
throw new IllegalStateException("No such any registry to reference");
}
// 这里省略了多个注册中心的情况
if (urls.size() == 1) {
// 我们通过注册中心生成了invoker，它代表一个可执行体，可向它发起invoke调用
invoker = refprotocol.refer(interfaceClass, urls.get(0));
// 这里生成invoker还包含了两个额外的步骤 1.consumer注册到registry 2.consumer订阅该服务下的providers/routers/configurators
// 另外这个invoker里包含了很多逻辑，比如cluster等
}
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");
}
if (logger.isInfoEnabled()) {
logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
}
// 创建服务代理
return (T) proxyFactory.getProxy(invoker);
}默认proxyFactory是StubProxyFactoryWrapper。从名字也能看出来，这个类里实现了客户端Stub的逻辑。不过这次我们就先跳过这一段，以后再来看。最终代理的实现无非还是JDK的动态代理或是Javassist的代理，我们比较关心InvocationHandler——InvokerInvocationHandler：
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);
}
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return invoker.toString();
}
if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
return invoker.hashCode();
}
if ("equals".equals(methodName) && parameterTypes.length == 1) {
return invoker.equals(args[0]);
}
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}其中RpcInvocation是会话域，它持有调用过程中的变量，比如方法名，参数等。
然后我们关注一下其中的成员变量invoker，最终的调用逻辑都是由invoker去执行。我们来看看这个invoker究竟是什么：
可以看到这里的invoker有点像装饰者模式，一层层的添加功能。需要注意的是，上述截图中包含MergeableClusterInvoker是因为我在reference上配置了group=*，而一般的流程则只会走：MockClusterInvoker -> FailoverClusterInvoker：其中 FailoverClusterInvoker 继承了 AbstractClusterInvoker，这个抽象类实现了 routers 的过滤 以及 loadBalance的过滤，最终会选取一个唯一的Provider来进行调用：
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
LoadBalance loadbalance;
// 这里的list方法也是根据directory的list方法进行过滤，也就是走routers进行过滤
List<Invoker<T>> invokers = list(invocation);
if (invokers != null && invokers.size() > 0) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
} else {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);
}核心类RegistryDirectory
这里提到了routers就不得不提到一个类——RegistryDirectory：每个ReferenceBean都会对应一个RegistryDirectory，其中维护了几乎所有关于该服务接口的配置：比如configurators、routers、providers等。
这个类实现了NotifyListener接口，显然每次当这些配置项有变化时都需要通知这个类做refresh。
其中每次刷新都会刷新这些配置项，并且还有一些缓存，比如：某个方法可调用的Provider的对应关系(methodInvokerMap)，简单看看这个notify方法：
public synchronized void notify(List<URL> urls) {
List<URL> invokerUrls = new ArrayList<URL>();
List<URL> routerUrls = new ArrayList<URL>();
List<URL> configuratorUrls = new ArrayList<URL>();
for (URL url : urls) {
String protocol = url.getProtocol();
String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
if (Constants.ROUTERS_CATEGORY.equals(category)
|| Constants.ROUTE_PROTOCOL.equals(protocol)) {
routerUrls.add(url);
} else if (Constants.CONFIGURATORS_CATEGORY.equals(category)
|| Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
configuratorUrls.add(url);
} else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
invokerUrls.add(url);
} else {
logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
}
}
// configurators
if (configuratorUrls != null && configuratorUrls.size() >0 ){
this.configurators = toConfigurators(configuratorUrls);
}
// routers
if (routerUrls != null && routerUrls.size() >0 ){
List<Router> routers = toRouters(routerUrls);
if(routers != null){ // null - do nothing
setRouters(routers);
}
}
List<Configurator> localConfigurators = this.configurators; // local reference
// 合并override参数
this.overrideDirectoryUrl = directoryUrl;
if (localConfigurators != null && localConfigurators.size() > 0) {
for (Configurator configurator : localConfigurators) {
this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
}
}
// providers
refreshInvoker(invokerUrls);
}这里会去设置routers、configurators等成员变量，然后在refereshInvoker方法中，重新构建各种缓存。
我们可以考虑一下该notify的触发点，肯定是当routers、configurators等信息有变化时才会触发。而这些信息都是存储在我们注册中心当中的，不得不提到Dubbo中最常见的注册中心——ZooKeeperRegistry
Registry
关于注册中心，在上面的分析中也提到过一次，不知道大家有没有注意到。就是在ReferenceConfig中createProxy方法中，有一步：
// 这里省略了多个注册中心的情况
if (urls.size() == 1) {
// 我们通过注册中心生成了invoker，它代表一个可执行体，可向它发起invoke调用
invoker = refprotocol.refer(interfaceClass, urls.get(0));
// 这里生成invoker还包含了两个额外的步骤 1.consumer注册到registry 2.consumer订阅该服务下的providers/routers/configurators
}这里我们可以深入再看看，这里我们用RegistryProtocol来做分析：
// 这里两个参数一个是服务类型，另一个是注册中心的url
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是注册中心的抽象，比如ZooKeeperRegistry
Registry registry = registryFactory.getRegistry(url);
if (RegistryService.class.equals(type)) {
return proxyFactory.getInvoker((T) registry, type, url);
}
// group="a,b" or group="*"
// 这里关于group一个和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);
}在doRefer方法中会生成Consumer的URL，并且将其注册到注册中心：
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);
URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
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)));
}
// 订阅该service下的 providers,configurators,routers
directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
Constants.PROVIDERS_CATEGORY
+ "," + Constants.CONFIGURATORS_CATEGORY
+ "," + Constants.ROUTERS_CATEGORY));
return cluster.join(directory);
}注册之后，就是消费者订阅，其中订阅了，一般都用ZK来做注册中心，利用ZK的watcher来做更新通知
protected void doSubscribe(final URL url, final NotifyListener listener) {
try {
if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {
// 省略部分代码
} else {
List<URL> urls = new ArrayList<URL>();
// 这里toCategoriesPath(url)返回的是一个该服务下routers等zk路径
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);
// 这里会返回该zk路径下的所有数据，比如provider的话，就返回多个provider的URL
List<String> children = zkClient.addChildListener(path, zkListener);
if (children != null) {
// 如果该path下的数据为空，那么返回empty://协议开头，暂时不知是何用意
// 注意：这里已经根据group等过滤过了
urls.addAll(toUrlsWithEmpty(url, path, children));
}
}
// 这里的listener也就是RegistryDirectory
notify(url, listener, urls);
}
} catch (Throwable e) {
throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
}
}订阅之后再执行一下notify，作为初始notify，之后每一次节点有变更，都会触发notify，而notify里的逻辑就是通知RegistryDirectory这个实例的内部配置的更新。其中第一个url是consumer的url，而第三个参数是更新的URL列表，注意，下面的notify方法是Registry中的notify，其中参数listener才是RegistryDirectory
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
if (url == null) {
throw new IllegalArgumentException("notify url == null");
}
if (listener == null) {
throw new IllegalArgumentException("notify listener == null");
}
if ((urls == null || urls.size() == 0)
&& ! Constants.ANY_VALUE.equals(url.getServiceInterface())) {
logger.warn("Ignore empty notify urls for subscribe url " + url);
return;
}
if (logger.isInfoEnabled()) {
logger.info("Notify urls for subscribe url " + url + ", urls: " + urls);
}
Map<String, List<URL>> result = new HashMap<String, List<URL>>();
for (URL u : urls) {
if (UrlUtils.isMatch(url, u)) {
String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
List<URL> categoryList = result.get(category);
if (categoryList == null) {
categoryList = new ArrayList<URL>();
result.put(category, categoryList);
}
categoryList.add(u);
}
}
if (result.size() == 0) {
return;
}
Map<String, List<URL>> categoryNotified = notified.get(url);
if (categoryNotified == null) {
notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
categoryNotified = notified.get(url);
}
for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
String category = entry.getKey();
List<URL> categoryList = entry.getValue();
categoryNotified.put(category, categoryList);
saveProperties(url);
listener.notify(categoryList);
}
}这里的url和listener应该是一一对应的，url代表一个consumer端的reference，而listener则代表该reference的RegistryDirectory，然后urls则代表多个配置项，这些配置项会一个个应用到RegistryDirectory上去。这里的urls根据category可以分成三类：routers、configurators、providers。并且每个category只会执行一次notify，之后每次这三个category的值有变更，都会进行通知，然后执行notify方法。
我们还可以来看看UrlUtils.isMatch(url, u)这个方法：
public static boolean isMatch(URL consumerUrl, URL providerUrl) {
String consumerInterface = consumerUrl.getServiceInterface();
String providerInterface = providerUrl.getServiceInterface();
if( ! (Constants.ANY_VALUE.equals(consumerInterface) || StringUtils.isEquals(consumerInterface, providerInterface)) ) return false;
if (! isMatchCategory(providerUrl.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY),
consumerUrl.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY))) {
return false;
}
if (! providerUrl.getParameter(Constants.ENABLED_KEY, true)
&& ! Constants.ANY_VALUE.equals(consumerUrl.getParameter(Constants.ENABLED_KEY))) {
return false;
}
String consumerGroup = consumerUrl.getParameter(Constants.GROUP_KEY);
String consumerVersion = consumerUrl.getParameter(Constants.VERSION_KEY);
String consumerClassifier = consumerUrl.getParameter(Constants.CLASSIFIER_KEY, Constants.ANY_VALUE);
String providerGroup = providerUrl.getParameter(Constants.GROUP_KEY);
String providerVersion = providerUrl.getParameter(Constants.VERSION_KEY);
String providerClassifier = providerUrl.getParameter(Constants.CLASSIFIER_KEY, Constants.ANY_VALUE);
return (Constants.ANY_VALUE.equals(consumerGroup) || StringUtils.isEquals(consumerGroup, providerGroup) || StringUtils.isContains(consumerGroup, providerGroup))
&& (Constants.ANY_VALUE.equals(consumerVersion) || StringUtils.isEquals(consumerVersion, providerVersion))
&& (consumerClassifier == null || Constants.ANY_VALUE.equals(consumerClassifier) || StringUtils.isEquals(consumerClassifier, providerClassifier));
}上面这段代码实现了group、version等之间的隔离逻辑。
OK，总算看完了一整个从Consumer调用远程服务的过程，具体的通信细节这里就先不展开讨论了（已经累得半死了）。
重回Routers
前面在分析源码时提到了RegistryDirectory，路由逻辑就是在这个类中完成的：
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
if (destroyed){
throw new RpcException("Directory already destroyed .url: "+ getUrl());
}
List<Invoker<T>> invokers = doList(invocation);
List<Router> localRouters = this.routers; // local reference
if (localRouters != null && localRouters.size() > 0) {
for (Router router: localRouters){
try {
if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, true)) {
invokers = router.route(invokers, getConsumerUrl(), invocation);
}
} catch (Throwable t) {
logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
}
}
}
return invokers;
}我们来细看一下上面的代码：
第一步——doList，这一步实际上是直接获取 方法和Provider的List的缓存关系
第二步——很关键，这里需要提到一个参数，就是这个runtime。每一条route都会有一个runtime这个属性，如果runtime配置为true，那么客户端会在每次调用服务端方法时都进行route一下，因为比如说根据参数来进行route，当然这对性能会有一定的影响，但是比如条件路由的话，也就正则表达式的判断，性能影响应该不是特别大。而runtime配置为false的话，就可以进行缓存，也就是在第一次调用时确定下来，后面只有当route规则进行变更了之后才会更新该缓存，也就是能直接在第一步中获得。这里又要吐槽一下官方文档了，童鞋们还记得前面的梗么。。。
再一次，继续吐槽，Router中有一个BUG，ConditionRouter中有两个核心方法，一个是matchWhen，另一个是MatchThen。也就是先判断Consumer是否满足matchWhen，若满足，则路由到满足matchThen条件的Provider
其中matchWhen方法：
private boolean matchCondition(Map<String, MatchPair> condition, URL url, URL param) {
Map<String, String> sample = url.toMap();
for (Map.Entry<String, String> entry : sample.entrySet()) {
String key = entry.getKey();
MatchPair pair = condition.get(key);
if (pair != null && ! pair.isMatch(entry.getValue(), param)) {
return false;
}
}
return true;
}而这个url.toMap();方法，得到的是Consumer的parameters的Map集合。我们可以通过注册中心观察到Consumer的URL上是没有method的参数的，类似如下：
consumer://192.168.1.88/com.alibaba.dubbo.examples.merge.api.MergeService?application=merge-consumer&category=providers,configurators,routers&dubbo=2.0.0&interface=com.alibaba.dubbo.examples.merge.api.MergeService&&pid=11020&side=consumer×tamp=1466450710697所以返回的map里是没有method这个key的，取而代之的是methods，而methods的value是所有的方法名，比如：methods=mergeResult,testRouter。
而官网上给我们的条件路由的配置里面有如下图所示，我又醉了。。。。
另外还有两点需要注意
1. 如果有多条routers规则，那么会根据每一条routers来过滤出可调用的provider列表
2. 针对条件路由，当经过某条路由规则路由后，没有一个符合规则的Provider，那么此次路由失败，会直接返回路由本条规则前的所有Provider，也就是相当于没有经过该路由的结果。
配置Routers
边看边分析边写，虽然精简了不少，但是还是多而乱，不知道是否有交代清楚。最后我们以实战配置一条routers来结束这篇文章。
场景
假设我们有多个服务分了不同的group，然后需要根据某个参数调用不同的group的方法
设计
先自定义一个Router，依据原有的ConditionRouter和ConditionRouterFactory创建出我们的CustomRouter和CoustomRouterFactory，当然核心逻辑还是在CustomRouter的route方法中，也很简单，只是根据请求参数路由一下而已：
public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
Object[] arguments = invocation.getArguments();
if(arguments == null || arguments.length == 0){
logger.error("method:" + invocation.getMethodName() + " do not have route param");
return null;
}
Object routeKey = arguments[0];
List<Invoker<T>> result = new ArrayList<Invoker<T>>();
for (Invoker<T> invoker : invokers) {
String group = invoker.getUrl().getParameter(Constants.GROUP_KEY);
if ((getGroup(routeKey)).equals(group)) {
result.add(invoker);
return result;
}
}
result.add(invokers.get(0));
return result;
}然后再让Dubbo自动发现我们添加的类：
custom=com.alibaba.dubbo.rpc.cluster.router.custom.CustomRouterFactory
最后再配置路由规则：
RegistryFactory registryFactory = ExtensionLoader.getExtensionLoader(RegistryFactory.class)
.getAdaptiveExtension();
Registry registry = registryFactory.getRegistry(URL.valueOf("zookeeper://127.0.0.1:2181"));
registry.register(URL.valueOf(
"routers://0.0.0.0/com.alibaba.dubbo.examples.merge.api.MergeService?name=test&category=routers&router=custom&dynamic=false"));这段配置我加在了Provider启动的时候。好了，最后模拟Consumer调用测试一下，OK
写在最后
总算写完了。已经好久没有写Blog了，接下来希望自己可以再重新坚持起来！加油
以上是关于由Dubbo路由引发的一言不合就看源码的主要内容，如果未能解决你的问题，请参考以下文章