Dubbo源代码分析七：使用executes属性的一个问题

我们知道，在Dubbo中可以给Provider配置线程池大小来控制系统提供服务的最大并行度，默认是200个，如果我们想配置成500，可以如下配置：

 

<dubbo:provider token="true" threads="500"/>

 

当我们想限制某个dubbo服务使用的最大线程数量时，dubbo提供了executes这一属性来提供这个功能，比如我们想限制某个接口最大能同时使用线程池中的100个线程，我们可以如下配置：

 

<dubbo:service interface="com.manzhizhen.service.MyLoverService" executes="100" />

 

我们看下dubbo内部executes是如何实现的，这就得移步到ExecuteLimitFilter，我们直接看下它的实现：

@Activate(group = Constants.PROVIDER, value = Constants.EXECUTES_KEY)

public class ExecuteLimitFilter implements Filter {

 

    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {

        URL url = invoker.getUrl();

        String methodName = invocation.getMethodName();

        int max = url.getMethodParameter(methodName, Constants.EXECUTES_KEY, 0);

        // 如果该接口/方法设置了executes并且值大于0

        if (max > 0) {

            // 取出该接口/方法对应的计数器

            RpcStatus count = RpcStatus.getStatus(url, invocation.getMethodName());

            // 如果当前使用的线程数量已经大于等于设置的阈值，那么直接抛出异常

            if (count.getActive() >= max) {

                throw new RpcException("Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ", cause: The service using threads greater than <dubbo:service executes=\"" + max + "\" /> limited.");

            }

        }

        long begin = System.currentTimeMillis();

        boolean isException = false;

        // 计数器+1

        RpcStatus.beginCount(url, methodName);

        try {

            Result result = invoker.invoke(invocation);

            return result;

        } catch (Throwable t) {

            isException = true;

            if (t instanceof RuntimeException) {

                throw (RuntimeException) t;

            } else {

                throw new RpcException("unexpected exception when ExecuteLimitFilter", t);

            }

        } finally {

            // 在finally中进行计数器-1

            RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, isException);

        }

    }

}

 

看上面的代码，可以得知基本步骤就是（黄底的部分代码）：计数器当前值和阈值比较 > 计数器+1 > 计数器-1。这种方式在高并发时会出现静态条件问题的，比如当前该接口已经使用了99个线程，这是时候有两个请求同时到达都发现count.getActive()是小于max的，于是该接口使用的线程数就有可能达到了101个。

 

那么，我们能不能把RpcStatus.beginCount(url, methodName);放到count.getActive() >= max的前面去执行？仔细想想后也不行，这样做的话有可能在高并发时请求被count.getActive() >= max卡死，因为大量请求将计数器+1（+1的速度远大于原有请求执行完将计数器-1的速度），导致一段时间内计数器一直大于阈值但实际上该接口使用的线程数却是0。

 

于是，为了将比较和+1做成原子的，我们想到了Semaphore，信号量Semaphore维护了一组许可，用来管理有限的资源，比如这里的线程数。使用Semaphore的有两点需要注意的地方，第一个就是说如果使用不当会导致Semaphore中的许可数多于最初设置的值，或者变为负数，也就是说，Semaphore并没有对非正常使用作出任何保护措施。还有一点是Semaphore没有方法能直接修改许可数量，但我们可以通过间接方法（比如多release几次以增加许可数量），但毕竟不优雅。

 

<!--StartFragment--> <!--EndFragment-->

Semaphore可以解决上述问题，但当需要修改线程数时，我们应该新建一个Semaphore对象，当采用这种替换Semaphore对象来应对许可数的变化时，一定要确保在仍和情况下acquire和release操作应该在一个Semaphore对象上。这也是我们在设计那些控制并发的小工具时需要注意的地方。

评论

5 楼 manzhizhen 2017-08-12  

yingwuluohan 写道

楼主，我还想请教个问题，一起分析下。
场景是几台台分布式部署的集群服务器，rpc通信用的dubbo，生产和消费也都按照dubbo框架的消费生产模式拆分的。目前运行的进程在上线几周后经常出现服务调用超时的情况，查询数据库监控并没有发现DB有锁表的情况，服务器cpu也正常。在重启应用服务器后超时的现象消失，应用又正常了。
  我几次想在问题发生时下载服务器的内存快照，想找出具体是应用的哪个对象占用了资源。不知道楼主熟悉java堆栈的监控的工具吗？能不能推荐一下，另外就是JVM的调优，当时也考虑到是不是GC没有及时回收导致的，我始终没有找到合适的工具分析内存快照，还有就是怎么样去权衡JVM的配置是否合适

先通过gc日志看是否是gc时间过长导致的超时，如果不是gc原因，可以通过请求日志来判断是否是网络耗时（consumer处理耗时-provider处理耗时）导致。如果不是gc和网络原因，看是否代码内部有什么锁或者队列导致的耗时增加，需要具体情况具体分析。

4 楼 yingwuluohan 2017-08-11  

楼主，我还想请教个问题，一起分析下。
场景是几台台分布式部署的集群服务器，rpc通信用的dubbo，生产和消费也都按照dubbo框架的消费生产模式拆分的。目前运行的进程在上线几周后经常出现服务调用超时的情况，查询数据库监控并没有发现DB有锁表的情况，服务器cpu也正常。在重启应用服务器后超时的现象消失，应用又正常了。
  我几次想在问题发生时下载服务器的内存快照，想找出具体是应用的哪个对象占用了资源。不知道楼主熟悉java堆栈的监控的工具吗？能不能推荐一下，另外就是JVM的调优，当时也考虑到是不是GC没有及时回收导致的，我始终没有找到合适的工具分析内存快照，还有就是怎么样去权衡JVM的配置是否合适

3 楼 yingwuluohan 2017-08-04  

manzhizhen 写道

yingwuluohan 写道

楼主，RpcStatus 中所有对计数的方法全部都是静态的，这样做与给方法加锁的效果是一样的吗？

静态方法和线程安全的方法是没有直接联系的。

我理解错了

2 楼 manzhizhen 2017-07-29  

yingwuluohan 写道

楼主，RpcStatus 中所有对计数的方法全部都是静态的，这样做与给方法加锁的效果是一样的吗？

静态方法和线程安全的方法是没有直接联系的。

1 楼 yingwuluohan 2017-07-28  

楼主，RpcStatus 中所有对计数的方法全部都是静态的，这样做与给方法加锁的效果是一样的吗？