在对数据库或者其它有限资源进行操作的时候，不能对这些资源无限制放任其增长，这个时候就需要使用流控方法来对资源的访问进行限制。下边都假设资源为数据库，其它资源也类似。
通常使用两种方式来对数据库进行限制：qps或者rt（response time）
个人认为qps限制适合资源可承受最高访问量稳定的时候使用，例如数据库可承受最高qps为1000，如果数据库
是应用独占，这个时候就可以限制应用对数据库的最高qps为1000，但是如果不是自己独占，例如还有另外一个应用同时在使用该数据库，这个时候就要调小本应用的最高限制，例如
500，太小浪费资源会浪费，太大会影响数据库稳定。
而利用rt时间来对访问进行限制，就可以实时调整对资源的访问量，例如默认qps最高为1000，设置最高rt不能大于1秒，当大于1秒的时候就需要调低对数据库的并发访问量。但是使用rt来流控时，队列是必不可少的组件，因为rt是实时计算的，当rt升高时，如果没有队列就会有大量的请求在这段时间进入数据库，结果可想而知。我认为这里边的队列有两种：同步队列和异步队列。
下边介绍一种自适应的流控设计，其中队列采用一个同步的基于timeslice的队列，其中有几个参数：
（1）qps_max：限制队列的最大大小，防止超过数据库可承受的最高限制
（2）timeslice的时间片大小：越小越准确，但是不能太小，默认不小于500毫秒
（3）Threshold：这个可以是一个时间范围，rt只要在这个范围内，就可以不更改qps大小（其中qps是实时根据rt来调整的，但是qps<qps_max）
下边第一段是从tddl复制的,相当于一个同步队列，第二段是实现简单的adaptive 流控，不完善但是可以作为参考
import java.util.concurrent.atomic.AtomicInteger;/** * 将时间片分为多个槽，每个槽一个计数器。游标按时间循环遍历每个槽。游标移动时才清零并且只清零当前的槽； * 因为采用mod计算(cursor = currentTime % timeslice/aSlotTime)，游标到头会自动折回来，事实上是一个环 * * cursor * | * +---------------------+-------------------------+ * | | | | | | C | | | | | | | * +-----------------------------------------------+ * | | * \-----------------timeslice-------------------/ * * * @author linxuan * */public class TimesliceFlowControl { private final static int MAX_SLOT = 20; //最多20片 private final static int MIN_SLOT_TIME = 500; //slot时间最少500毫秒 private final String name; private final AtomicInteger[] slots; //槽数组，最小的时间粒度数组 private final int aSlotTimeMillis; //一个槽的时间，最小的时间单位 private final int timesliceMillis; //总的时间窗口（时间片）大小 private final int timesliceMaxIns; //时间片内允许的最大访问次数(进入个数) private final AtomicInteger total = new AtomicInteger(); //总的计数 private final AtomicInteger totalReject = new AtomicInteger(); //总的拒绝/超限计数 private volatile int cursor = 0; //游标 private volatile long cursorTimeMillis = System.currentTimeMillis(); //当前slot的开始时间 /** * @param name 流控的名称 * @param aSlotTimeMillis //一个槽的时间 * @param slotCount //槽的数目 * @param timesliceMaxIns //时间窗口内最多允许执行的次数，设为0则不限制 */ public TimesliceFlowControl(String name, int aSlotTimeMillis, int slotCount, int timesliceMaxIns) { if (slotCount < 2) { throw new IllegalArgumentException("slot至少要有两个"); } this.name = name; this.aSlotTimeMillis = aSlotTimeMillis; this.timesliceMillis = aSlotTimeMillis * slotCount; this.timesliceMaxIns = timesliceMaxIns; slots = new AtomicInteger[slotCount]; for (int i = 0; i < slotCount; i++) { slots[i] = new AtomicInteger(0); } } /** * 最小的时间单位取默认的500毫秒 * @param name 流控的名称 * @param timesliceMillis 时间片; 传0表示使用默认值1分钟 * @param timesliceMaxIns 时间片内最多允许执行多少次，设为0则不限制 */ public TimesliceFlowControl(String name, int timesliceMillis, int timesliceMaxIns) { if (timesliceMillis == 0) { timesliceMillis = 60 * 1000; //时间片默认1分钟 } if (timesliceMillis < 2 * MIN_SLOT_TIME) { throw new IllegalArgumentException("time slice less than" + (2 * MIN_SLOT_TIME)); } //this(name, 500, timesliceMillis / 500, limit); int slotCount = MAX_SLOT; //默认分20个slot int slotTime = timesliceMillis / slotCount; if (slotTime < MIN_SLOT_TIME) { slotTime = MIN_SLOT_TIME; //如果slot时间小于MIN_SLOT_TIME，则最小半秒 slotCount = timesliceMillis / slotTime; } this.name = name; this.aSlotTimeMillis = slotTime; //this.timesliceMillis = timesliceMillis; //直接赋值因为截余的关系，会数组越界 this.timesliceMillis = aSlotTimeMillis * slotCount; this.timesliceMaxIns = timesliceMaxIns; slots = new AtomicInteger[slotCount]; for (int i = 0; i < slotCount; i++) { slots[i] = new AtomicInteger(0); } } public void check() { if (!allow()) { throw new IllegalStateException(reportExceed()); } } public String reportExceed() { return name + " exceed the limit " + timesliceMaxIns + " in timeslice " + timesliceMillis; } public boolean allow() { final long current = System.currentTimeMillis(); final int index = (int) ((current % timesliceMillis) / aSlotTimeMillis); if (index != cursor) { int oldCursor = cursor; cursor = index; //尽快赋新值 final long oldCursorTimeMillis = cursorTimeMillis; cursorTimeMillis = current; //尽快赋新值 //多个线程会进入下面这里，但是每个线程计算的total会大致相同 if (current - oldCursorTimeMillis > timesliceMillis) { //时间差大于timesliceMillis，则整个时间片都应该清零了 for (int i = 0; i < slots.length; i++) { slots[i].set(0); //清零，忽略并发造成的计数出入 } this.total.set(0); } else { do { //吃尾（尾清零），考虑跳跃的情况 oldCursor++; if (oldCursor >= slots.length) { oldCursor = 0; } slots[oldCursor].set(0); //清零，忽略并发造成的计数出入 } while (oldCursor != index); //int clearCount = slots[index].get(); //slots[index].set(0); //清零，忽略并发造成的计数出入 int newtotal = 0; for (int i = 0; i < slots.length; i++) { newtotal += slots[i].get(); //包括了新的当前槽 } this.total.set(newtotal); //设置总数，忽略并发造成的计数出入 } } else { if (current - cursorTimeMillis > aSlotTimeMillis) { //index相同但是时间差大于一个slot的时间，说明整个时间片都需要清零了 cursorTimeMillis = current; //尽快赋新值 for (int i = 0; i < slots.length; i++) { slots[i].set(0); //清零，忽略并发造成的计数出入 } this.total.set(0); } //是否为了避免开销，不做上面的判断？ } if (timesliceMaxIns == 0) { return true; //0为不限制 } if (this.total.get() < timesliceMaxIns) { //放来的才计数，拒绝的不计数 slots[index].incrementAndGet(); total.incrementAndGet(); return true; } else { totalReject.incrementAndGet(); return false; } } /** * @return 当前时间片内的总执行次数 */ public int getCurrentCount() { return total.get(); } /** * @return 返回有史以来(对象创建以来)被拒绝/超限的次数 */ public int getTotalRejectCount() { return totalReject.get(); }} import java.util.concurrent.atomic.AtomicReference;/** * Email: diwayou@163.com * User: diwayou * Date: 13-5-14 * Time: 上午12:35 */public class AdaptiveFlowControl { private static final float QPS_UPDATE_PARAM = 1.2f; // 合适？ private final int qpsMax; // qps上限 private final int qpsMin; // qps下限 private final int rtThresholdStart; // 阈值范围值start 单位毫秒 private final int rtThresholdEnd; // 阈值范围值end 单位毫秒 private final String name; private volatile int qps; private final AtomicReference timesliceFlowControl; public AdaptiveFlowControl(String name, int qps, int rtThresholdStart, int rtThresholdEnd, int qpsMax, int qpsMin) { this.qps = qps; this.rtThresholdEnd = rtThresholdEnd; this.rtThresholdStart = rtThresholdStart; this.qpsMax = qpsMax; this.qpsMin = qpsMin; this.name = name; this.timesliceFlowControl = new AtomicReference (new TimesliceFlowControl(name, 0, qps)); } public boolean allow() { return timesliceFlowControl.get().allow(); } public void update(int newRt) { if (newRt < rtThresholdStart) { int newQps = (int)(qps * QPS_UPDATE_PARAM); if (newQps > qpsMax) newQps = qpsMax; this.timesliceFlowControl.set(new TimesliceFlowControl(name, 0, newQps)); } else if (newRt > rtThresholdEnd) { int newQps = (int)(qps / QPS_UPDATE_PARAM); if (newQps < qpsMin) newQps = qpsMin; this.timesliceFlowControl.set(new TimesliceFlowControl(name, 0, newQps)); } }} 参考论文：
[1]
[2]
[3]
[4]