「DUBBO系列」集群容错之Failover

<beans>
  <dubbo:application name="xpz-consumer" >
  <dubbo:registry address="zookeeper://127.0.0.1:2181" >
  <dubbo:reference id="helloService" cluster="failover" retries="5" interface="com.itxpz.dubbo.demo.provider.HelloService" >
</beans>
复制

public class Consumer {
    public static void main(String[] args) throws Exception {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[] { "classpath*:META-INF/spring/dubbo-consumer.xml" });
        context.start();
        HelloService helloService = (HelloService) context.getBean("helloService");
        String result = helloService.sayHello("IT徐胖子");
        System.out.println("==========客户端收到结果==========" + result);
    }
}
复制

// 通过动态代理机制生成代理对象Proxy
// Proxy持有InvokerInvocationHandler
// InvokerInvocationHandler持有invoker = MockClusterInvoker(FailoverClusterInvoker)
HelloService helloService = (HelloService) context.getBean("helloService");
复制

public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
   
    public FailoverClusterInvoker(Directory<T> directory) {
        super(directory);
    }


    @Override
    public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        
        // 所有生产者invokers
        List<Invoker<T>> copyInvokers = invokers;
        checkInvokers(copyInvokers, invocation);
        String methodName = RpcUtils.getMethodName(invocation);


        // 获取重试次数
        int len = getUrl().getMethodParameter(methodName, Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
        if (len <= 0) {
            len = 1;
        }
        RpcException le = null;


        // 已经调用过的生产者
        List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyInvokers.size());
        Set<String> providers = new HashSet<String>(len);


        // 重试直到达到最大次数
        for (int i = 0; i < len; i++) {
            if (i > 0) {


                // 如果当前实例被销毁则抛出异常
                checkWhetherDestroyed();


                // 根据路由策略选出可用生产者Invokers
                copyInvokers = list(invocation);


                // 重新检查
                checkInvokers(copyInvokers, invocation);
            }


            // 负载均衡选择一个生产者invoker
            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 " + methodName + " 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()) {
                    throw e;
                }
                le = e;
            } catch (Throwable e) {
                le = new RpcException(e.getMessage(), e);
            } finally {
                providers.add(invoker.getUrl().getAddress());
            }
        }
        throw new RpcException(le.getCode(), "Failed to invoke the method " + methodName + " in the service " + getInterface().getName() + ". Tried " + len + " times of the 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.getCause() != null ? le.getCause() : le);
    }
}
复制