某个流控工具分析

   

  处理流程：

1.       在web容器启动时，通过SphFilter启动降级线程，初始化整个Sph框架。

Web.xml配置:

<filter>
     <filter-name>SphFilter</filter-name>
     <display-name>SphFilter</display-name>
     <filter-class>com.xxxx.common.stable.SphFilter</filter-class>
  </filter>
  <filter-mapping>
     <filter-name>SphFilter</filter-name>
     <url-pattern>*.htm</url-pattern>
  </filter-mapping>
<filter-mapping>

 

 

Init代码:

     ……
       // load default conf
       SphDRuntime.init(null);

        if (monitor != null) {
           monitor.stop();
       }
       monitor = newSphMonitor();
       new Thread(monitor).start();
   }

 

2.       业务代码初始化CtSph

Sph sph = new CtSph("GetBuyerLevel", 10, 50,ValveType.COUNT_AND_AVGELAPSED_VALVE_TYPE);

 

3.       业务代码调用CtSph.entry()

1)       判断是否被流控

if(valveType == ValveType.COUNT_VALVE_TYPE || valveType ==ValveType.COUNT_AND_AVGELAPSED_VALVE_TYPE) {
           isEntry = (count.get() < countValve.get()) ? true : false;
        }

 

比较两个AtomicInteger大小，非线程安全，高并发时会有高于阀值线程进入，作者说不用CAS是因为性能，并发数没有精确控制需求

2)       初始化ThreadLocal的entries变量，Entry的作用是保存线程的进入时间，支持可重入，同一个线程可以多次使用同一个流控组件，最多100层，用数组模拟stack

3)       如果不流控，则递增当前并发数，否，输出日志

 

4.       业务代码运行一段时间后，调用CsSph.release

1)       Entries递减，释放掉stack(用数组模拟)上最顶层的entry，记录消耗时间，并将时间存入统计数组ConcurrentCircleArray

2)       递减当前并发数

 

后台降级线程SphMonitor逻辑:

1.       遍历SphDRuntime中的每个流控组件

2.       计算该组件的平均消耗时间

int aveElapsed = sph.getDetailElapsed().getArrayAvg();

3.       如果超过阀值则降级，这里有个最低并发值保证，默认为0

                                                              if (aveElapsed > sph.getAvgElpasedValve()) {
                               if (sph.getCountValve().get() <=sph.getLowCountValve()) {
                                   log.info(key + "LowCountValve=" + sph.getLowCountValve());
                               } else {
                                   int cv = sph.getCountValve().addAndGet(-2);
                                   log.info(key + " decrementCountValve=" + cv + " avgElapsed=" +aveElapsed);
                               }
                           }

 

4.       如果没超过阀值，则恢复并发阀值

                      if (sph.getCountValve().get() <sph.getDefaultCountValve()) {
                                   sph.getCountValve().set(sph.getDefaultCountValve());
                                   log.info(key + " resetCountValve=" + sph.getDefaultCountValve());
                               }

 

评论

1 楼 hewei1997 2013-07-21  

讲了代码，但是没讲原理，我觉得其实原理更重要一些