Dubbo学习笔记12:使用Dubbo中需要注意的一些事情

摘要:
此外,如果异步调用的方法没有返回值,则可以调用methodConfig。setReturn;再一次减少在Dubbo中创建和管理未来对象的成本。此外,应注意,如果check设置为false,则将始终返回调用。当服务提供商恢复服务时,它可以自动连接。服务使用者ReferenceConfig需要缓存ReferenceConfig实例本身,这是一个重实例。每个ReferenceConfig实例都与服务注册表和所有服务提供者保持一个长链。特别是,在使用Dubbo API编程时,很容易忽略这个问题。

指定方法异步调用

前面我们讲解了通过设置ReferenceConfig的setAsync()方法来让整个接口里的所有方法变为异步调用,那么如何指定某些方法为异步调用呢?下面讲解下如何正确地设置默写方法为异步调用。

假如你只需要设置接口里的方法sayHello为异步调用,那么可以使用下面方式:

final List<MethodConfig> asyncMethodList = new ArrayList<MethodConfig>();
MethodConfig methodConfig = new MethodConfig();
methodConfig.setAsync(true);
methodConfig.setName("sayHello");
asyncMethodList.add(methodConfig);

然后调用ReferenceConfig的setMethods(asyncMethodList)即可。另外如果异步调用的方法没有返回值,则可以再调用methodConfig.setReturn(false); ,以便减少Dubbo内部Future对象的创建和管理成本。

关闭启动检查

正常情况下,在服务消费端启动时会检查依赖的服务是否存在,如果不存在则会抛出 throw new IllegalStateException("Failed to check the status of the service" + interfaceName + ".No provider available for the service ")异常阻止Spring初始化完成,以便上线前能及早发现问题。

可以通过调用ReferenceConfig.setCheck(false)关闭检查,设置check为true有助于及时发现问题,那么什么时候需要设置false呢?

比如测试时,有些无关的服务启动不了,导致整个应用都启动不了,这时候你可以把那些无关服务的check设置为false。再比如出现了循环依赖,必须有一方先启动,比如你给服务使用方提供了一个SDK,SDK里面使用Dubbo API远程消费服务器提供方的服务,如果你在服务提供方的服务器里面也引入这个SDK,在服务提供方启动时候就会抛出 No Provider available for the service异常,具体原因是服务提供方启动时候会初始化SDK,而SDK里面初始化时候需要检查服务提供方是否存在,而服务提供方的服务还没提供出来。

另外需要注意的是check设置为false,总是会返回调用,当服务提供方恢复服务时,能自动连上。

如何设置均衡策略

由于Dubbo提供的一致性Hash负载均衡策略,可以允许你自定义虚拟节点个数和指定某些方法需要使用一致性Hash策略,下面具体讲下如何设置:

复制代码
// 虚拟节点设置为512
Map<String,String> parameters = new HashMap<String,String>();
parameters.put("hash.nodes","512");
ReferenceConfig<T> reference = new ReferenceConfig<T>();
// 设置负载均衡为一致性Hash
reference.setLoadbalance(consistenthash);
// 设置参数
reference.setParameters(parameters);
复制代码

如下代码设置接口的sayHello方法为一致性Hash负载均衡策略,设置saySomething方法为随机负载均衡策略:

复制代码
ReferenceConfig reference = new ReferenceConfig();
final List<MethodConfig> methodList = new ArrayList<MethodConfig>();
// 设置sayHello方法为一致性Hash负载均衡策略
MethodConfig methodConfig = new MethodConfig();
methodConfig.setName("sayHello");
methodConfig.setLoadbalance("consistenthash");
// 虚拟节点设置为512
Map<String,String> parameters = new HashMap<String,String>();
parameters.put("hash.nodes","512");
methodConfig.setParameters(parameters);
methodList.add(methodConfig);
// 设置saySomething方法为随机负载均衡策略
methodConfig = new MethodConfig();
methodConfig.setName("saySomething");
methodConfig.setLoadbalance("random");
methodList.add(methodConfig);
reference.setMethods(methodList);
复制代码

另外,默认情况下一致性hash使用第一个参数值计算hash值,如果你需要自定义可以通过以下设置:

Map<String,String> parameters = new HashMap<String,String>();
parameters.put("hash.nodes","512");
parameters.put("hash.arguments","0,1");    // 使用第一个和第二个参数值计算hash值
methodConfig.setParameters(parameters);
    

注意"0,1"是一个字符串,里面使用英文","分隔。

服务消费端ReferenceConfig需要自行缓存

ReferenceConfig实例是个很重的实例,每个ReferenceConfig实例里都维护了与服务注册中心的一个长链,并且维护了与所有服务提供者的长链。假设有一个服务注册中心和N个服务提供者,那么每个ReferenceConfig实例里面维护了N+1个长链,如果频繁地生成ReferenceConfig实例,可能会造成性能问题,甚至产生内存或者连接泄露的风险。特别是使用Dubbo API编程时候容易忽略这个问题。

为了解决这个问题,之前都是自行缓存,但自从发布Dubbo 2.4.0版本后,Dubbo提供了简单的工具类ReferenceConfigCache用于缓存ReferenceConfig实例。使用如下:

复制代码
// 创建服务消费实例
ReferenceConfig<XxxService> reference = new ReferenceConfig<XxxService>();
reference.setInterface(XxxService.class);
reference.setVersion("1.0.0");
......
// 获取Dubbo提供的缓存
ReferenceConfigCache cache = ReferenceConfigCache.getCache();
// cache.get方法中会缓存reference对象,并且调用reference.get方法启动ReferenceConfig,并返回经过代理后的服务接口的对象
XxxService xxxService = cache.get(reference);
// 使用xxxService对象
xxxService.sayHello();     
复制代码

需要注意的是Cache内持有ReferenceConfig对象的引用,不要在外部再调用ReferenceConfig的destroy方法了,这会导致Cache内的ReferenceConfig失效!

如果要销毁Cache中的ReferenceConfig,将销毁ReferenceConfig并释放对应的资源,具体使用下面方法来销毁:

ReferenceConfigCache cache = ReferenceConfigCache.getCache();
cache.destroy(reference);

另外在Dubbo中唯一确定一个服务是通过 接口+分组+版本,所以默认情况下cache内是通过服务Group/接口/版本三个属性来标示一个ReferenceConfig实例的。即以服务Group/接口/版本为缓存的key,ReferenceConfig实例为对应的value。如果你需要使用自定义的key,可以在创建cache时候调用 ReferenceConfigCache cache = ReferenceConfigCache.getCache(keyGenerator); 方法传递自定义的keyGenerator。

并发控制

服务消费方并发控制

在服务消费方进行并发控制,需要设置actives参数,如下:

<dubbo:reference   interface="com.test.UserServiceBo" group="dubbo" version="1.0.0" timeout="3000" actives="10" />

设置com.test.UserServiceBo接口中的所有方法,每个方法最多同时并发请求10个请求。

也可以使用下面方法设置接口中单个方法的并发请求个数,如下:

<dubbo:reference   interface="com.test.UserServiceBo" group="dubbo" version="1.0.0" timeout="3000">
    <dubbo:method name="sayHello" actives="10" />
</dubbo:reference>

如上设置sayHello方法的并发请求数量最大为10,如果客户端请求该方法并发超过了10则客户端会被阻塞,等客户端并发请求数量少于10的时候,该请求才会被发送到服务提供方服务器。在Dubbo中客户端并发控制使用ActiveLimitFilter过滤器来控制,代码如下:

复制代码
public class ActiveLimitFilter implements Filter{
    public Result invoke(Invoker<?> invoker , Invocation invocation) throws RpcException{
        URL url = invoker.getUrl();
        String methodName = invocation.getMethodName();
        // 获取设置的active值,默认为0
        int max = invoker.getUrl().getMethodParameter(methodName , Constants.ACTIVES_KEY , 0);
        RpcStatus count = RpcStatus.getStatus(invoker.getUrl() , invocation.getMethodName());
        if(max > 0){
            long timeout = invoker.getUrl().getMethodParameter(invocation.getMethodName() , Constants.TIMEOUT_KEY , 0);
            long start = System.currentTimeMillis();
            long remain = timeout;
            int active = count.getActive();
            // 如果该方法并发请求数量大于设置值,则挂起当前线程
            if(active >= max){
                sychronized(count){
                    while((active = count.getActive()) >= max){
                        try{
                            count.wait(remain);
                        }catch(InterruptedException e){
                            
                        }
                        // 如果等待时间超时,则抛出异常
                        long elapsed = System.currentTimeMillis() - start;
                        remain = timeout - elapsed;
                        if(remain <= 0){
                            throw new RpcException("Waiting concurrent invoke timeout in client-side for service: " + invoker.getInterface().getName() + " , method: " + invocation.getMethodName() + ",elapsed: " + elapsed + ",timeout: " + timeout + ". concurrent invokes: " + active + ". max concurrent invoke limit:" + max);
                        }
                    }
                }
            }

        }
        // 没有限流的时候,正常调用
        try{
            long begin = System.currentTimeMillis();
            RpcStatus.beginCount(url , methodName);
            try{
                Result result = invoker.invoke(invocation);
                RpcStatus.endCount(url , methodName , System.currentTimeMillis() - begin , true);
                return result;
            }catch(RuntimeException t){
                RpcStatus.endCount(url , methodName , System.currentTimeMillis() - begin , false);
                throw t;
            }
        }finally{
            if(max > 0){
                synchronized(count){
                    count.notify();
                }
            }
        }
    }
}
复制代码

由上可知,在客户端并发控制中,如果当并发量达到指定值后,当前客户端请求线程会被挂起,如果在等待超时期间并发请求量少了,那么阻塞的线程会被激活,然后发送请求到服务提供方,如果等待超时了,则直接抛出异常,这时服务根本都没有发送到服务提供方服务器。

服务提供方并发控制

在服务提供方进行并发控制需要设置executes参数,如下:

<dubbo:service interface="com.test.UserServiceBo" ref="userService" group="dubbo" version="1.0.0" timeout="3000" executes="10" />

设置com.test.UserServiceBo 接口中所有方法,每个方法最多同时并发处理10个请求,这里并发是指同时在处理10个请求。

也可以使用下面方法设置接口中单个方法的并发处理个数,如下:

<dubbo:service interface="com.test.UserServiceBo" ref="userService" group="dubbo" version="1.0.0" timeout="3000">
    <dubbo:method name="sayHello" executes="10" />
</dubbo:service>

如上设置sayHello方法的并发处理数量为10 。

需要注意的是,服务提供方设置并发数量后,如果同时请求数量大于了设置的executes的值,则会抛出异常,而不是像服务端设置actives时候,会等待。服务提供方并发控制是使用ExecuteLimitFilter过滤器实现的,ExecuteLimitFilter代码如下:

复制代码
public class ExecuteLimitFilter implements Filter{
    public Result invoke(Invoker<?> invoker , Invocation invocation) throws RpcException{
        URL url = invoker.getUrl();
        // 默认不设置executes时候,其值为0
        int max = url.getMethodParameter(methodName , Constants.EXECUTES_KEY , 0);
        if(max > 0){    // max>0说明设置了executes值
            RpcStatus count = RpcStatus.getStatus(url , invocation.getMethodName());
            // 可知如果并发处理数量大于设置的值,会抛出异常
            executesLimit = count.getSemaphore(max);
            if(executesLimit != null && !(acquireResult = executesLimit.tryAcquire())){
                throw new RpcException("Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ",cause: The service using threads greater than <dubbo:service execute="" + max + ""> limited.");
            }
        }
        ...
        try{    // 没有限流的时候,激活filter链
            Result result = invoker.invoke(invocation);
            return result;
        }catch(){
            ...
        }finally{
            ...
        }

    }
}
复制代码

所以当使用executes参数时候要注意,当并发量过大时侯,多余的请求会失败。

改进的广播策略

前面我们讲解集群容错时谈到广播策略,该策略主要用于对所有服务提供者广播消息,那么有个问题需要思考,广播是说你在客户端调用接口一次,内部就是轮询调用所有服务提供者的机器的服务,那么你调用一次该接口,返回值是什么呢?比如内部轮询了10台机器,每个机器应该都有一个返回值,那么你调用的这一次返回值是10个返回值的组成?其实不是,返回的轮询调用的最后一个机器结果,我们可以看下BroadcastClusterInvoker的主干代码:

复制代码
public class BroadcastClusterInvoker<T> extends AbstractClusterInvoker<T>{
    private static final Logger logger = LoggerFactory.getLogger(BroadcastClusterInvoker.class);

    public BroadcastClusterInvoker(Directory<T> directory){
        super(directory);
    }

    @SuppressWarnings({"unchecked","rawtypes"})
    public Result doInvoke(final Invocation invocation , List<Invoker<T>> invokers , LoadBalance loadbalance){
        ...
        // 使用循环,轮询每个机器进行调用,其中result为最后一个机器的结果
        for(Invoker<T> invoker : invokers){
            try{
                result = invoker.invoke(invocation);
            }catch(RpcException e){
                exception = e;
                logger.warn(e.getMessage() , e);
            }catch(Throwable e){
                exception = new RpcException(e.getMessage(),e);
                logger.warn(e.getMessage() , e);
            }
        }
        if(exception != null){
            throw exception;
        }
        return result;
    }
}        
复制代码

如上代码,可知使用循环轮询调用每个机器,其中result为调用最后一个机器的结果。

如果我想获取所有服务提供者的结果,该怎么办呢?其实我们可以自定义一个SPI扩展,并且规定我们的服务接口的返回结果为一个map,代码如下:

复制代码
public Result doInvokePara(final Invocation invocation , List<Invoker<T>> invokers , LoadBalance loadbalance) throws RpcException{
    // 用来保存所有服务提供者返回的结果
    Map allResult = new ConcurrentHashMap<String , Result>();
    // 保存异步调用返回的Future对象
    List<Future<Result>> futureList = new ArrayList<Future<Result>>();
    // 所有服务提供者的机器个数
    int machineNum = invokers.size();
    for(Invoker<T> invoker : invokers){
        try{
            // 异步调用服务提供者
            Future<Result> future = paramCallPool.submit(new Callable<Result>(){
                @Override
                public Result call() throws Exception{
                    try{
                        // 具体调用服务提供者
                        Result result = invoker.invoke(invocation);
                        // 服务提供者 ip:port
                        String url = invoker.getUrl().getAddress();
                        // 保存结果到map,key为服务提供者的地址,value为返回结果
                        allResult.put(url , result.getResult());
                        return result;
                    }catch(RpcException e){
                        logger.warn(e.getMessage(),e);
                    }catch(Throwable e){
                        logger.warn(e.getMessage(),e);
                    }
                    return null;
                }
            });
            futureList.add(future);
        }catch(Exception e){
            logger.warn(e.getMessage() , e);
        }
    }
    // 等所有调用完成
    for(Future<Result> future : futureList){
        try{
            future.get();
        }catch(InterruptedException | ExecutionException e){
            e. printStackTrace();   
        }
    }

    // 假设服务接口返回中类型为这个
    ACCSResult<Map> resultTemp = new ActionResult<Map>(true,null,null,null);
    // 自定义返回结果
    Map finalResult = new HashMap<String,Result>();
    finalResult.put("machineNum",machineNum);
    finalResult.put("result",result);
    resultTemp.setData(finalResult);
    // 重新设置返回值
    Result result = new RpcResult(resultTemp);
    return result;    
}
复制代码

免责声明:文章转载自《Dubbo学习笔记12:使用Dubbo中需要注意的一些事情》仅用于学习参考。如对内容有疑问,请及时联系本站处理。

上篇kafka2.5.0 主题Topicwindows 中如何定位恶意软件的藏身位置下篇

宿迁高防,2C2G15M,22元/月;香港BGP,2C5G5M,25元/月 雨云优惠码:MjYwNzM=

相关文章

log4net.redis+logstash+kibana+elasticsearch+redis 实现日志系统

前端时间写了个随笔 log4net.NoSql +ElasticSearch 实现日志记录 ,因项目原因需要把日志根java平台的同事集成采用logstash+kibana+elasticsearch+redis结构实现日志统计分析,所以需要一个将log4net日志输出到redis的组件。没有找到现成的,就自己动手了。参考了 log4net.NoSql的代...

SpringBoot 动态修改定时任务频率

原文:https://cloud.tencent.com/developer/article/1627714 import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import or...

Java的泛型

版权声明:本文系博主原创,未经博主许可,禁止转载。保留所有权利。 引用网址:https://www.cnblogs.com/zhizaixingzou/p/9992847.html 目录 1. 泛型1.1. 解决什么问题 先来看如下的一个类,它封装了一个String类型的值。 1 public class StringEntry { 2 p...

【java开发系列】—— java输入输出流

前言   任何语言输入输出流都是很重要的部分,比如从一个文件读入内容,进行分析,或者输出到另一个文件等等,都需要文件流的操作。这里简单介绍下reader,wirter,inputstream,outputstream的使用方法。其实Apache commons里面有个方法IOUtils可是实现方便快捷的流拷贝,感兴趣的可以参考官方文档。   JAVA的...

Android 演示 Android ListView 和 github XListView(1-3)

本文内容 环境 项目结构 演示 1:ListView 演示 2:简单 XListView 演示 3:音乐列表 XListView 演示 4:另一个音乐列表 XListView 本文四个演示,循序渐进。 演示 1 只是普通的 Android ListView 控件; 演示 2 是 GitHub 上的 XListView 控件,具备“下拉...

mybatis-plus学习

mybatis-plus 一、简介 通过maven项目,springBoot构建学习,官方地址文档简介 | MyBatis-Plus (baomidou.com) 官方讲解,mybatis的优点: 无侵入:只做增强不做改变,引入它不会对现有工程产生影响,如丝般顺滑 损耗小:启动即会自动注入基本 CURD,性能基本无损耗,直接面向对象操作 强大的 CRUD...