From bd90e94f4325d3f387d2b4e7459a898961e231a5 Mon Sep 17 00:00:00 2001 From: tydhot <583125614@qq.com> Date: Mon, 28 Dec 2020 01:08:22 +0800 Subject: [PATCH 1/3] add Sentinel LongAdder --- README.md | 1 + ...41\346\225\260\345\256\236\347\216\260.md" | 61 +++++++++++++++++++ 2 files changed, 62 insertions(+) create mode 100644 "docs/Sentinel/Sentinel\345\272\225\345\261\202LongAdder\347\232\204\350\256\241\346\225\260\345\256\236\347\216\260.md" diff --git a/README.md b/README.md index 3a5e0a88..9c64d861 100644 --- a/README.md +++ b/README.md @@ -255,6 +255,7 @@ ## Sentinel - [sentinel 时间窗口实现](docs/Sentinel/Sentinel时间窗口的实现.md) +- [Sentinel 底层LongAdder的计数实现](docs/Sentinel/Sentinel底层LongAdder的计数实现.md) ## 番外篇（JDK 1.8） diff --git "a/docs/Sentinel/Sentinel\345\272\225\345\261\202LongAdder\347\232\204\350\256\241\346\225\260\345\256\236\347\216\260.md" "b/docs/Sentinel/Sentinel\345\272\225\345\261\202LongAdder\347\232\204\350\256\241\346\225\260\345\256\236\347\216\260.md" new file mode 100644 index 00000000..31a96bfd --- /dev/null +++ "b/docs/Sentinel/Sentinel\345\272\225\345\261\202LongAdder\347\232\204\350\256\241\346\225\260\345\256\236\347\216\260.md" @@ -0,0 +1,61 @@ +## LongAdder 的原理 +在 LongAdder 中，底层通过多个数值进行累加来得到最后的结果。当多个线程对同一个 LongAdder 进行更新的时候，将会对这一些列的集合进行动态更新，以避免多线程之间的资源竞争。当需要得到 LongAdder 的具体的值的时候，将会将一系列的值进行求和作为最后的结果。 +在高并发的竞争下进行类似指标数据的收集的时候，LongAdder 通常会和 AtomicLong 进行比较，在低竞争的场景下，两者有着相似的性能表现。而当在高并发竞争的场景下，LongAdder 将会表现更高的性能，但是也会伴随更高的内存消耗。 +## LongAdder 的代码实现 + +```java + transient volatile Cell[] cells; + transient volatile long base; +``` + +cells 是一个简单的 Cell 数组，当比如通过 LongAdder 的 add()方法进行 LongAdder 内部的数据的更新的时候，将会根据每个线程的一个 hash 值与 cells 数组的长度进行取模而定位，并在定位上的位置进行数据更新。而 base 则是当针对 LongAdder 的数据的更新时，并没有线程竞争的时候，将会直接更新在 base 上，而不需要前面提到的 hash 再定位过程，当 LongAdder 的 sum()方法被调用的时候，将会对 cells 的所有数据进行累加在加上 sum 的值进行返回。 + +```java + public long sum() { + long sum = base; + Cell[] as = cells; + if (as != null) { + int n = as.length; + for (int i = 0; i < n; ++i) { + Cell a = as[i]; + if (a != null) { sum += a.value; } + } + } + return sum; + } +``` + +相比 sum()方法，LongAdder 的 add()方法要复杂得多。 + +```java + public void add(long x) { + Cell[] as; + long b, v; + HashCode hc; + Cell a; + int n; + if ((as = cells) != null || !casBase(b = base, b + x)) { + boolean uncontended = true; + int h = (hc = threadHashCode.get()).code; + if (as == null || (n = as.length) < 1 || + (a = as[(n - 1) & h]) == null || + !(uncontended = a.cas(v = a.value, v + x))) { retryUpdate(x, hc, uncontended); } + } + } +``` + +在 add()方法的一开始，将会观察 cells 数组是否存在，如果不存在，将会尝试直接通过 casBase()方法在 base 上通过 cas 更新，这是在低并发竞争下的 add()流程，这一流程的前提是对于 LongAdder 的更新并没有遭遇别的线程的并发修改。 +在当 cells 已经存在，而或者对于 base 的 cas 更新失败，都将会将数据的更新落在 cells 数组之上。首先，每个线程都会在其 ThreadLocal 中生成一个线程专有的随机数，并根据这个随机数与 cells 进行取模，定位到的位置进行 cas 修改。在这个流程下，由于根据线程专有的随机数进行 hash 而定位的流程，尽可能的避免了线程间的资源竞争。但是仍旧可能存在 hash 碰撞而导致两个线程定位到了同一个 cells 槽位的情况，这里就需要通过 retryUpdate()方法进行进一步的解决。 +retryUpdate()方法的代码很长，但是逻辑很清晰，主要分为一下几个流程，其中的主流程是一个死循环，进入 retryUpdate()方法后，将会不断尝试执行主要逻辑，直到对应的逻辑执行完毕： +1.当进入 retryUpdate()的时候，cells 数组还没有创建，将会尝试获取锁并初始化 cells 数组并直接在 cells 数组上进行修改，而别的线程在没创建的情况下进入并获取锁失败，将会直接尝试在 base 上进行更行。 2.当进入 retryUpdate()的时候，cells 数组已经创建，但是分配给其的数组槽位的 Cells 还没有进行初始化，那么将会尝试获取锁并对该槽位进行初始化。 3.当进入 retryUpdate()的时候，cells 数组已经创建，分配给其的槽位的 Cell 也已经完成了初始化，而是因为所定位到的槽位与别的线程发生了 hash 碰撞，那么将会加锁并扩容 cells 数组，之后对该线程持有的 hash 进行 rehash，在下一轮循环中对新定位的槽位数据进行更新。而别的线程在尝试扩容并获取锁失败的时候，将会直接对自己 rehash 并在下一轮的循环中重新在新的 cells 数组中进行定位更新。 +## Cell 本身的内存填充 +最后，提一下 cells 数组中的 Cell 对象。 + +```java + volatile long p0, p1, p2, p3, p4, p5, p6; + volatile long value; + volatile long q0, q1, q2, q3, q4, q5, q6; +``` + +每个 Cell 对象中具体存放的 value 前后都由 7 个 long 类型的字段进行内存填充以避免缓存行伪共享而导致的缓存失效。 + From 25aee0f85467d0be4d21e8e8006211caa56626f8 Mon Sep 17 00:00:00 2001 From: Yang Libin Date: Mon, 28 Dec 2020 01:16:53 +0800 Subject: [PATCH 2/3] =?UTF-8?q?Update=20Sentinel=E5=BA=95=E5=B1=82LongAdde?= =?UTF-8?q?r=E7=9A=84=E8=AE=A1=E6=95=B0=E5=AE=9E=E7=8E=B0.md?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...41\346\225\260\345\256\236\347\216\260.md" | 63 ++++++++++--------- 1 file changed, 33 insertions(+), 30 deletions(-) diff --git "a/docs/Sentinel/Sentinel\345\272\225\345\261\202LongAdder\347\232\204\350\256\241\346\225\260\345\256\236\347\216\260.md" "b/docs/Sentinel/Sentinel\345\272\225\345\261\202LongAdder\347\232\204\350\256\241\346\225\260\345\256\236\347\216\260.md" index 31a96bfd..e6b268b4 100644 --- "a/docs/Sentinel/Sentinel\345\272\225\345\261\202LongAdder\347\232\204\350\256\241\346\225\260\345\256\236\347\216\260.md" +++ "b/docs/Sentinel/Sentinel\345\272\225\345\261\202LongAdder\347\232\204\350\256\241\346\225\260\345\256\236\347\216\260.md" @@ -1,61 +1,64 @@ ## LongAdder 的原理 + 在 LongAdder 中，底层通过多个数值进行累加来得到最后的结果。当多个线程对同一个 LongAdder 进行更新的时候，将会对这一些列的集合进行动态更新，以避免多线程之间的资源竞争。当需要得到 LongAdder 的具体的值的时候，将会将一系列的值进行求和作为最后的结果。 -在高并发的竞争下进行类似指标数据的收集的时候，LongAdder 通常会和 AtomicLong 进行比较，在低竞争的场景下，两者有着相似的性能表现。而当在高并发竞争的场景下，LongAdder 将会表现更高的性能，但是也会伴随更高的内存消耗。 +在高并发的竞争下进行类似指标数据的收集的时候，LongAdder 通常会和 AtomicLong 进行比较，在低竞争的场景下，两者有着相似的性能表现。而当在高并发竞争的场景下，LongAdder 将会表现更高的性能，但是也会伴随更高的内存消耗。 + ## LongAdder 的代码实现 ```java - transient volatile Cell[] cells; - transient volatile long base; +transient volatile Cell[] cells; +transient volatile long base; ``` cells 是一个简单的 Cell 数组，当比如通过 LongAdder 的 add()方法进行 LongAdder 内部的数据的更新的时候，将会根据每个线程的一个 hash 值与 cells 数组的长度进行取模而定位，并在定位上的位置进行数据更新。而 base 则是当针对 LongAdder 的数据的更新时，并没有线程竞争的时候，将会直接更新在 base 上，而不需要前面提到的 hash 再定位过程，当 LongAdder 的 sum()方法被调用的时候，将会对 cells 的所有数据进行累加在加上 sum 的值进行返回。 ```java - public long sum() { - long sum = base; - Cell[] as = cells; - if (as != null) { - int n = as.length; - for (int i = 0; i < n; ++i) { - Cell a = as[i]; - if (a != null) { sum += a.value; } - } +public long sum() { + long sum = base; + Cell[] as = cells; + if (as != null) { + int n = as.length; + for (int i = 0; i < n; ++i) { + Cell a = as[i]; + if (a != null) { sum += a.value; } } - return sum; } + return sum; +} ``` 相比 sum()方法，LongAdder 的 add()方法要复杂得多。 ```java - public void add(long x) { - Cell[] as; - long b, v; - HashCode hc; - Cell a; - int n; - if ((as = cells) != null || !casBase(b = base, b + x)) { - boolean uncontended = true; - int h = (hc = threadHashCode.get()).code; - if (as == null || (n = as.length) < 1 || - (a = as[(n - 1) & h]) == null || - !(uncontended = a.cas(v = a.value, v + x))) { retryUpdate(x, hc, uncontended); } - } +public void add(long x) { + Cell[] as; + long b, v; + HashCode hc; + Cell a; + int n; + if ((as = cells) != null || !casBase(b = base, b + x)) { + boolean uncontended = true; + int h = (hc = threadHashCode.get()).code; + if (as == null || (n = as.length) < 1 || + (a = as[(n - 1) & h]) == null || + !(uncontended = a.cas(v = a.value, v + x))) { retryUpdate(x, hc, uncontended); } } +} ``` 在 add()方法的一开始，将会观察 cells 数组是否存在，如果不存在，将会尝试直接通过 casBase()方法在 base 上通过 cas 更新，这是在低并发竞争下的 add()流程，这一流程的前提是对于 LongAdder 的更新并没有遭遇别的线程的并发修改。 在当 cells 已经存在，而或者对于 base 的 cas 更新失败，都将会将数据的更新落在 cells 数组之上。首先，每个线程都会在其 ThreadLocal 中生成一个线程专有的随机数，并根据这个随机数与 cells 进行取模，定位到的位置进行 cas 修改。在这个流程下，由于根据线程专有的随机数进行 hash 而定位的流程，尽可能的避免了线程间的资源竞争。但是仍旧可能存在 hash 碰撞而导致两个线程定位到了同一个 cells 槽位的情况，这里就需要通过 retryUpdate()方法进行进一步的解决。 retryUpdate()方法的代码很长，但是逻辑很清晰，主要分为一下几个流程，其中的主流程是一个死循环，进入 retryUpdate()方法后，将会不断尝试执行主要逻辑，直到对应的逻辑执行完毕： 1.当进入 retryUpdate()的时候，cells 数组还没有创建，将会尝试获取锁并初始化 cells 数组并直接在 cells 数组上进行修改，而别的线程在没创建的情况下进入并获取锁失败，将会直接尝试在 base 上进行更行。 2.当进入 retryUpdate()的时候，cells 数组已经创建，但是分配给其的数组槽位的 Cells 还没有进行初始化，那么将会尝试获取锁并对该槽位进行初始化。 3.当进入 retryUpdate()的时候，cells 数组已经创建，分配给其的槽位的 Cell 也已经完成了初始化，而是因为所定位到的槽位与别的线程发生了 hash 碰撞，那么将会加锁并扩容 cells 数组，之后对该线程持有的 hash 进行 rehash，在下一轮循环中对新定位的槽位数据进行更新。而别的线程在尝试扩容并获取锁失败的时候，将会直接对自己 rehash 并在下一轮的循环中重新在新的 cells 数组中进行定位更新。 + ## Cell 本身的内存填充 + 最后，提一下 cells 数组中的 Cell 对象。 ```java - volatile long p0, p1, p2, p3, p4, p5, p6; - volatile long value; - volatile long q0, q1, q2, q3, q4, q5, q6; +volatile long p0, p1, p2, p3, p4, p5, p6; +volatile long value; +volatile long q0, q1, q2, q3, q4, q5, q6; ``` 每个 Cell 对象中具体存放的 value 前后都由 7 个 long 类型的字段进行内存填充以避免缓存行伪共享而导致的缓存失效。 - From 4e3d5915395bc4e5099ca45cf8b2be6d4b0d14f4 Mon Sep 17 00:00:00 2001 From: tydhot <583125614@qq.com> Date: Wed, 13 Jan 2021 00:16:03 +0800 Subject: [PATCH 3/3] add Sentinel RateLimiter --- README.md | 1 + ...25\347\232\204\345\256\236\347\216\260.md" | 140 ++++++++++++++++++ 2 files changed, 141 insertions(+) create mode 100644 "docs/Sentinel/Sentinel\351\231\220\346\265\201\347\256\227\346\263\225\347\232\204\345\256\236\347\216\260.md" diff --git a/README.md b/README.md index 9c64d861..54232e11 100644 --- a/README.md +++ b/README.md @@ -256,6 +256,7 @@ - [sentinel 时间窗口实现](docs/Sentinel/Sentinel时间窗口的实现.md) - [Sentinel 底层LongAdder的计数实现](docs/Sentinel/Sentinel底层LongAdder的计数实现.md) +- [Sentinel 限流算法的实现](docs/Sentinel/Sentinel限流算法的实现.md) ## 番外篇（JDK 1.8） diff --git "a/docs/Sentinel/Sentinel\351\231\220\346\265\201\347\256\227\346\263\225\347\232\204\345\256\236\347\216\260.md" "b/docs/Sentinel/Sentinel\351\231\220\346\265\201\347\256\227\346\263\225\347\232\204\345\256\236\347\216\260.md" new file mode 100644 index 00000000..e84e4ec8 --- /dev/null +++ "b/docs/Sentinel/Sentinel\351\231\220\346\265\201\347\256\227\346\263\225\347\232\204\345\256\236\347\216\260.md" @@ -0,0 +1,140 @@ +## Sentinel 中漏桶算法的实现 + +Sentinel 中漏桶算法通过 RateLimiterController 来实现，在漏桶算法中，会记录上一个请求的到达时间，如果新到达的请求与上一次到达的请求之间的时间差小于限流配置所规定的最小时间，新到达的请求将会排队等待规定的最小间隔到达，或是直接失败。 + +```Java +@Override + public boolean canPass(Node node, int acquireCount, boolean prioritized) { + if (acquireCount <= 0) { + return true; + } + + if (count <= 0) { + return false; + } + + long currentTime = TimeUtil.currentTimeMillis(); + // 根据配置计算两次请求之间的最小时间 + long costTime = Math.round(1.0 * (acquireCount) / count * 1000); + + // 计算上一次请求之后，下一次允许通过的最小时间 + long expectedTime = costTime + latestPassedTime.get(); + + if (expectedTime <= currentTime) { + // 如果当前时间大于计算的时间，那么可以直接放行 + latestPassedTime.set(currentTime); + return true; + } else { + // 如果没有，则计算相应需要等待的时间 + long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis(); + if (waitTime > maxQueueingTimeMs) { + return false; + } else { + long oldTime = latestPassedTime.addAndGet(costTime); + try { + waitTime = oldTime - TimeUtil.currentTimeMillis(); + // 如果最大等待时间小于需要等待的时间，那么返回失败，当前请求被拒绝 + if (waitTime > maxQueueingTimeMs) { + latestPassedTime.addAndGet(-costTime); + return false; + } + // 在并发条件下等待时间可能会小于等于0 + if (waitTime > 0) { + Thread.sleep(waitTime); + } + return true; + } catch (InterruptedException e) { + } + } + } + return false; + } +``` + +## Sentinel 中令牌桶算法的实现 + +在 Sentinel 中，令牌桶算法通过 WarmUpController 类实现。在这个情况下，当配置每秒能通过多少请求后，那么在这里 sentinel 也会每秒往桶内添加多少的令牌。当一个请求进入的时候，将会从中移除一个令牌。由此可以得出，桶内的令牌越多，也说明当前的系统利用率越低。因此，当桶内的令牌数量超过某个阈值后，那么当前的系统可以称之为处于`饱和`状态。 +当系统处于 `饱和`状态的时候，当前允许的最大 qps 将会随着剩余的令牌数量减少而缓慢增加，达到为系统预热热身的目的。 + +```Java + this.count = count; + + this.coldFactor = coldFactor; + + warningToken = (int)(warmUpPeriodInSec * count) / (coldFactor - 1); + + maxToken = warningToken + (int)(2 * warmUpPeriodInSec * count / (1.0 + coldFactor)); + + slope = (coldFactor - 1.0) / count / (maxToken - warningToken); +``` + +其中 count 是当前 qps 的阈值。coldFactor 则为冷却因子，warningToken 则为警戒的令牌数量，warningToken 的值为(热身时间长度 _ 每秒令牌的数量) / (冷却因子 - 1)。maxToken 则是最大令牌数量，具体的值为 warningToken 的值加上 (2 _ 热身时间长度 _ 每秒令牌数量) / (冷却因子 + 1)。当当前系统处于热身时间内，其允许通过的最大 qps 为 1 / (超过警戒数的令牌数 _ 斜率 slope + 1 / count)，而斜率的值为(冷却因子 - 1) / count / (最大令牌数 - 警戒令牌数)。 +举个例子: count = 3， coldFactor = 3，热身时间为 4 的时候，警戒令牌数为 6，最大令牌数为 12，当剩余令牌处于 6 和 12 之间的时候，其 slope 斜率为 1 / 9。 那么当剩余令牌数为 9 的时候的允许 qps 为 1.5。其 qps 将会随着剩余令牌数的不断减少而直到增加到 count 的值。 + +```Java + @Override + public boolean canPass(Node node, int acquireCount, boolean prioritized) { + long passQps = (long) node.passQps(); + + long previousQps = (long) node.previousPassQps(); + // 首先重新计算其桶内剩余的数量 + syncToken(previousQps); + + // 开始计算它的斜率 + // 如果进入了警戒线，开始调整他的qps + long restToken = storedTokens.get(); + if (restToken >= warningToken) { + long aboveToken = restToken - warningToken; + // 如果当前剩余的令牌数大于警戒数，那么需要根据准备的计算公式重新计算qps，这个qps小于设定的阈值 + double warningQps = Math.nextUp(1.0 / (aboveToken * slope + 1.0 / count)); + if (passQps + acquireCount <= warningQps) { + return true; + } + } else { + if (passQps + acquireCount <= count) { + return true; + } + } + + return false; + } + + protected void syncToken(long passQps) { + long currentTime = TimeUtil.currentTimeMillis(); + currentTime = currentTime - currentTime % 1000; + long oldLastFillTime = lastFilledTime.get(); + if (currentTime <= oldLastFillTime) { + return; + } + + long oldValue = storedTokens.get(); + long newValue = coolDownTokens(currentTime, passQps); + + if (storedTokens.compareAndSet(oldValue, newValue)) { + // 从桶内移除相应数量的令牌，并更新最后更新时间 + long currentValue = storedTokens.addAndGet(0 - passQps); + if (currentValue < 0) { + storedTokens.set(0L); + } + lastFilledTime.set(currentTime); + } + + } + + private long coolDownTokens(long currentTime, long passQps) { + long oldValue = storedTokens.get(); + long newValue = oldValue; + + // 当令牌的消耗程度远远低于警戒线的时候，将会补充令牌数 + if (oldValue < warningToken) { + newValue = (long)(oldValue + (currentTime - lastFilledTime.get()) * count / 1000); + } else if (oldValue > warningToken) { + if (passQps < (int)count / coldFactor) { + // qps小于阈值 / 冷却因子的时候，说明此时还不需要根据剩余令牌数调整qps的阈值，所以也会补充 + newValue = (long)(oldValue + (currentTime - lastFilledTime.get()) * count / 1000); + } + } + return Math.min(newValue, maxToken); + } +``` +