2018/5/18 15:03:53当前位置推荐好文程序员浏览文章

一、前言

前面讲解了Dubbo的服务降级,本节我们来讲解dubbo中的并发控制,并发控制分为客户端并发控制和服务端并发控制。

二、并发控制

2.1 客户端并发控制

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

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

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

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

    <dubbo:reference id="userService" 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();        //获取设置的acvites的值,默认为0        int max = invoker.getUrl().getMethodParameter(methodName, Constants.ACTIVES_KEY, 0);        //获取当前方法目前并发请求数量        RpcStatus = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName());        if (max > 0) {//说明设置了actives变量            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) {                synchronized (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();                }            }        }    }}

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

2.2 服务端并发控制

在服务提供方进行并发控制需要设置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 = 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 executes=\"" + max + "\" /> limited.");                }            }            ...            try {//没有限流时候,激活filter链                Result result = invoker.invoke(invocation);                return result;            } catch (Throwable t) {             ...            } finally {               ...            }        }    }

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

三、总结

本节我们讲解了dubbo中客户端并发控制和服务端并发控制。另外另外想系统学dubbo的单击我 ,想学并发的童鞋能 单击我

网友评论