dubbo 服务降级_大数据系统

dubbo 服务降级

官网：https://dubbo.apache.org/zh/docs/advanced/local-mock/

服务降级

应用：消费端服务调用失败，不抛出异常，返回默认的数据

服务降级配置示例

# 同目录下查找服务降级类：com.foo.service.BarServiceMock


# 指定本地服务降级类：com.foo.service.impl.BarServiceMock


# 方法级别服务容错：方法名后面加".mock"

mock 可选值

true：开启服务降级
false：不开启服务降级

# return：返回一个字符串表示的对象
empty：基本类型返回对应的默认值、集合类返回空值，如：return empty
null：返回null
true：返回true
false：返回false
JSON字符串：返回json字符串反序列化后的对象


# throw：抛出异常对象
抛出rpcException： 
抛出自定义异常：

# force：强制使用mock，不发起服务调用，可与return、throw组合使用
# fail：服务调用失败时，执行mock（默认行为），可与return、throw组合使用
强制返回fake：
强制抛出异常：

消费端应用启动时，在创建failoverClusterInvoker的过程中，会用MockClusterWrapper包装，调用栈如下：

# 创建FailoverClusterInvoker对象，默认创建该对象
at org.apache.dubbo.rpc.cluster.support.FailoverClusterInvoker.(FailoverClusterInvoker.java:52)
 
# join操作
at org.apache.dubbo.rpc.cluster.support.FailoverCluster.doJoin(FailoverCluster.java:33)
at org.apache.dubbo.rpc.cluster.support.wrapper.AbstractCluster.join(AbstractCluster.java:58)

# MockClusterWrapper：服务降级包装类
at org.apache.dubbo.rpc.cluster.support.wrapper.MockClusterWrapper.join(MockClusterWrapper.java:39)
 
# 创建invoker
at org.apache.dubbo.registry.integration.RegistryProtocol.doCreateInvoker(RegistryProtocol.java:564)
 
# 获取invoker
at org.apache.dubbo.registry.integration.InterfaceCompatibleRegistryProtocol.getInvoker(InterfaceCompatibleRegistryProtocol.java:58)
 
# MigrationInvoker类
at org.apache.dubbo.registry.client.migration.MigrationInvoker.refreshInterfaceInvoker(MigrationInvoker.java:448)
at org.apache.dubbo.registry.client.migration.MigrationInvoker.migrateToApplicationFirstInvoker(MigrationInvoker.java:239)
 
# MigrationRuleHandler类
at org.apache.dubbo.registry.client.migration.MigrationRuleHandler.refreshInvoker(MigrationRuleHandler.java:73)
at org.apache.dubbo.registry.client.migration.MigrationRuleHandler.doMigrate(MigrationRuleHandler.java:57)
	  - locked <0x21db> (a org.apache.dubbo.registry.client.migration.MigrationRuleHandler)
at org.apache.dubbo.registry.client.migration.MigrationRuleListener.onRefer(MigrationRuleListener.java:241)
 
# RegistryProtocol类
at org.apache.dubbo.registry.integration.RegistryProtocol.interceptInvoker(RegistryProtocol.java:531)
at org.apache.dubbo.registry.integration.RegistryProtocol.doRefer(RegistryProtocol.java:500)
at org.apache.dubbo.registry.integration.RegistryProtocol.refer(RegistryProtocol.java:485)
 
# refer操作
at org.apache.dubbo.rpc.protocol.ProtocolListenerWrapper.refer(ProtocolListenerWrapper.java:74)
at org.apache.dubbo.qos.protocol.QosProtocolWrapper.refer(QosProtocolWrapper.java:83)
at org.apache.dubbo.rpc.cluster.filter.ProtocolFilterWrapper.refer(ProtocolFilterWrapper.java:71)
at org.apache.dubbo.rpc.protocol.ProtocolSerializationWrapper.refer(ProtocolSerializationWrapper.java:52)
at org.apache.dubbo.rpc.Protocol$Adaptive.refer(Protocol$Adaptive.java:-1)
 
# ReferenceConfig类
at org.apache.dubbo.config.ReferenceConfig.createInvokerForRemote(ReferenceConfig.java:481)
at org.apache.dubbo.config.ReferenceConfig.createProxy(ReferenceConfig.java:386)
at org.apache.dubbo.config.ReferenceConfig.init(ReferenceConfig.java:275)
	  - locked <0x21dc> (a org.apache.dubbo.config.ReferenceConfig)
at org.apache.dubbo.config.ReferenceConfig.get(ReferenceConfig.java:216)
 
# SimpleReferenceCache类
at org.apache.dubbo.config.utils.SimpleReferenceCache.get(SimpleReferenceCache.java:110)
 
# DefaultModuleDeployer类
at org.apache.dubbo.config.deploy.DefaultModuleDeployer.lambda$referServices$6(DefaultModuleDeployer.java:384)
at org.apache.dubbo.config.deploy.DefaultModuleDeployer$$Lambda$880/0x0000000801039dc8.accept(Unknown Source:-1)
	  at java.util.concurrent.ConcurrentHashMap$ValuesView.forEach(ConcurrentHashMap.java:4780)
 
# DefaultModuleDeployer类
at org.apache.dubbo.config.deploy.DefaultModuleDeployer.referServices(DefaultModuleDeployer.java:364)
at org.apache.dubbo.config.deploy.DefaultModuleDeployer.start(DefaultModuleDeployer.java:151)
	  - locked <0x21dd> (a org.apache.dubbo.config.deploy.DefaultModuleDeployer)
 
# DubboDeployApplicationListener类
at org.apache.dubbo.config.spring.context.DubboDeployApplicationListener.onContextRefreshedEvent(DubboDeployApplicationListener.java:108)
at org.apache.dubbo.config.spring.context.DubboDeployApplicationListener.onApplicationEvent(DubboDeployApplicationListener.java:98)
at org.apache.dubbo.config.spring.context.DubboDeployApplicationListener.onApplicationEvent(DubboDeployApplicationListener.java:44)
 
# SimpleApplicationEventMulticaster类
at org.springframework.context.event.SimpleApplicationEventMulticaster.doInvokeListener(SimpleApplicationEventMulticaster.java:176)
at org.springframework.context.event.SimpleApplicationEventMulticaster.invokeListener(SimpleApplicationEventMulticaster.java:169)
at org.springframework.context.event.SimpleApplicationEventMulticaster.multicastEvent(SimpleApplicationEventMulticaster.java:143)
 
# AbstractApplicationContext类
at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:421)
at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:378)
at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:938)
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:586)
	  - locked <0x21de> (a java.lang.Object)
 
# ServletWebServerApplicationContext类
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:145)
 
# SpringApplication类
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:730)
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:412)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:302)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1301)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1290)
 
# 项目启动入口
at com.example.demo.DemoApplication.main(DemoApplication.java:12)

MockClusterWrapper：服务降级包装类

public class MockClusterWrapper implements Cluster {

    private final Cluster cluster;

    public MockClusterWrapper(Cluster cluster) {
        this.cluster = cluster;
    }

    @Override
    public  Invoker join(Directory directory, boolean buildFilterChain) throws RpcException {
        return new MockClusterInvoker(directory,
                this.cluster.join(directory, buildFilterChain));
    }   //返回MockClusterInvoker对象

    public Cluster getCluster() {
        return cluster;
    }
}

MockClusterInvoker：发起服务调用

public class MockClusterInvoker implements ClusterInvoker {

    private static final Logger logger = LoggerFactory.getLogger(MockClusterInvoker.class);

    private final Directory directory;
    private final Invoker invoker;

    public MockClusterInvoker(Directory directory, Invoker invoker) {
        this.directory = directory;
        this.invoker = invoker;
    }

    public URL getUrl() {
    public URL getRegistryUrl() {
    public Class getInterface() {
    public Directory getDirectory() {

    public void destroy() {
    public boolean isDestroyed() {
    public boolean isAvailable() {


    @Override
    public Result invoke(Invocation invocation) throws RpcException {  //服务调用
        Result result;

        String value = getUrl().getMethodParameter(invocation.getMethodName(), MOCK_KEY, Boolean.FALSE.toString()).trim();
                       //value：获取mock值，如果mock没有设置，默认为false
        if (value.length() == 0 || "false".equalsIgnoreCase(value)) {
            //no mock
            result = this.invoker.invoke(invocation);  //value长度为0或者值为false，直接发起服务调用，不使用服务降级，如果调用失败会抛出异常
        } else if (value.startsWith(FORCE_KEY)) {      //如果value以force开头，不发起远程调用，直接服务降级
            if (logger.isWarnEnabled()) {
                logger.warn("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + getUrl());
            }
            //force:direct mock
            result = doMockInvoke(invocation, null);   //服务降级操作
        } else {
            //fail-mock
            try {                                      //其他先执行服务调用，调用失败则执行服务降级
                result = this.invoker.invoke(invocation);   //执行远程调用

                //fix:#4585
                if(result.getException() != null && result.getException() instanceof RpcException){
                    RpcException rpcException= (RpcException)result.getException();
                    if(rpcException.isBiz()){
                        throw  rpcException;
                    }else {
                        result = doMockInvoke(invocation, rpcException);
                    }
                }

            } catch (RpcException e) {
                if (e.isBiz()) {      //如果是业务异常，直接抛出异常，不进行服务降级
                    throw e;
                }

                if (logger.isWarnEnabled()) {
                    logger.warn("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + getUrl(), e);
                }
                result = doMockInvoke(invocation, e);  //服务调用失败，进行服务降级
            }
        }
        return result;     //返回正常结果或者服务降级结果
    }

    @SuppressWarnings({"unchecked", "rawtypes"})
    private Result doMockInvoke(Invocation invocation, RpcException e) {
                           //服务降级
        Result result;
        Invoker mockInvoker;

        List> mockInvokers = selectMockInvoker(invocation);  //选择服务降级invoker
        if (CollectionUtils.isEmpty(mockInvokers)) {
            mockInvoker = (Invoker) new MockInvoker(getUrl(), directory.getInterface());
        } else {
            mockInvoker = mockInvokers.get(0);
        }   //有多个mockInvoker，选择第一个mockInvoker
        try {
            result = mockInvoker.invoke(invocation);   //服务降级调用
        } catch (RpcException mockException) {
            if (mockException.isBiz()) {
                result = AsyncRpcResult.newDefaultAsyncResult(mockException.getCause(), invocation);
            } else {
                throw new RpcException(mockException.getCode(), getMockExceptionMessage(e, mockException), mockException.getCause());
            }
        } catch (Throwable me) {
            throw new RpcException(getMockExceptionMessage(e, me), me.getCause());
        }
        return result;
    }

    private String getMockExceptionMessage(Throwable t, Throwable mt) {
        String msg = "mock error : " + mt.getMessage();
        if (t != null) {
            msg = msg + ", invoke error is :" + StringUtils.toString(t);
        }
        return msg;
    }

    
    private List> selectMockInvoker(Invocation invocation) {
        List> invokers = null;
        //TODO generic invoker？
        if (invocation instanceof RpcInvocation) {
            //Note the implicit contract (although the description is added to the interface declaration, but extensibility is a problem. The practice placed in the attachment needs to be improved)
            invocation.setAttachment(INVOCATION_NEED_MOCK, Boolean.TRUE.toString());
            //directory will return a list of normal invokers if Constants.INVOCATION_NEED_MOCK is absent or not true in invocation, otherwise, a list of mock invokers will return.
            try {
                RpcContext.getServiceContext().setConsumerUrl(getUrl());
                invokers = directory.list(invocation);
            } catch (RpcException e) {
                if (logger.isInfoEnabled()) {
                    logger.info("Exception when try to invoke mock. Get mock invokers error for service:"
                            + getUrl().getServiceInterface() + ", method:" + invocation.getMethodName()
                            + ", will construct a new mock with 'new MockInvoker()'.", e);
                }
            }
        }
        return invokers;
    }

    @Override
    public String toString() {
        return "invoker :" + this.invoker + ",directory: " + this.directory;
    }
}

使用示例

***********

provider

application.yml

dubbo:
  application:
    name: dubbo-provider
    register-mode: instance
  registry:
    address: zookeeper://localhost:2181
    group: dubbo
    #register-mode: instance
    #register: false
  protocol:
    name: dubbo
    port: 20880

HelloService

public interface HelloService {

    String hello();
}

HelloService2

public interface HelloService2 {

    String hello2();
}

HelloServiceImpl

@DubboService    //服务暴露
public class HelloServiceImpl implements HelloService {

    @Override
    public String hello() {
        System.out.println("hello provider");

        return "hello provider";
    }
}

HelloService2Impl

@DubboService    //服务暴露
public class HelloService2Impl implements HelloService2 {

    @Override
    public String hello2() {
        

        throw new RpcException("hello2 调用出错了");
    }   //抛出rpcException，消费端调用时执行服务降级操作
}

DemoApplication

@EnableDubbo    //扫描注册dubbo服务
@SpringBootApplication
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }

}

***********

consumer

application.yml

dubbo:
  application:
    name: dubbo-consumer
  registry:
    address: zookeeper://localhost:2181
    group: dubbo
    #register-mode: instance
  protocol:
    name: dubbo
    #port: 20880

server:
  port: 8081

HelloService

public interface HelloService {

    String hello();
}

HelloService2

public interface HelloService2 {

    String hello2();
}

HelloService2Mock

public class HelloService2Mock implements HelloService2 {

    @Override
    public String hello2() {
        return "hello2 mock";
    }
}

HelloController

@RestController
public class HelloController {

    @DubboReference(cluster = ClusterRules.FORKING, parameters = {"forks","3"})
    private HelloService helloService;

    @DubboReference(mock = "true")
    private HelloService2 helloService2;

    @RequestMapping("/hello")
    public String hello(){
        System.out.println(helloService.hello());
        System.out.println(helloService2.hello2());

        return "hello consumer 2";
    }
}

DemoApplication

@EnableDubbo       //开启dubbo
@SpringBootApplication
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }

}

***********

使用测试

localhost:8081/hello，控制台输出：

2022-02-11 12:26:18.615  INFO 1570 --- [nio-8081-exec-1] o.s.web.servlet.DispatcherServlet        : Initializing Servlet 'dispatcherServlet'
2022-02-11 12:26:18.615  INFO 1570 --- [nio-8081-exec-1] o.s.web.servlet.DispatcherServlet        : Completed initialization in 0 ms
hello provider
hello2 mock

helloService2服务调用失败，执行服务降级操作

dubbo 服务降级

大数据系统相关栏目本月热门文章