feat: 算法补充#1539

src/backend/common/common-ratelimiter/src/main/kotlin/com/tencent/bkrepo/common/ratelimiter/algorithm/DistributedLeakyRateLimiter.kt

@@ -0,0 +1,73 @@
+/*
+ * Tencent is pleased to support the open source community by making BK-CI 蓝鲸持续集成平台 available.
+ *
+ * Copyright (C) 2022 THL A29 Limited, a Tencent company.  All rights reserved.
+ *
+ * BK-CI 蓝鲸持续集成平台 is licensed under the MIT license.
+ *
+ * A copy of the MIT License is included in this file.
+ *
+ *
+ * Terms of the MIT License:
+ * ---------------------------------------------------
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of
+ * the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
+ * LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+ * NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+ * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+ * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package com.tencent.bkrepo.common.ratelimiter.algorithm
+
+import com.tencent.bkrepo.common.ratelimiter.exception.AcquireLockFailedException
+import com.tencent.bkrepo.common.ratelimiter.redis.LuaScript
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+import org.springframework.data.redis.core.RedisTemplate
+import org.springframework.data.redis.core.script.DefaultRedisScript
+import kotlin.system.measureTimeMillis
+
+/**
+ * 分布式漏桶算法实现
+ */
+class DistributedLeakyRateLimiter(
+    private val key: String,
+    private val permitsPerSecond: Double,
+    private val capacity: Long,
+    private val redisTemplate: RedisTemplate<String, String>,
+) : RateLimiter {
+    override fun tryAcquire(permits: Long): Boolean {
+        try {
+            var acquireResult = false
+            val elapsedTime = measureTimeMillis {
+                val redisScript = DefaultRedisScript(LuaScript.leakyRateLimiterScript, List::class.java)
+                val results = redisTemplate.execute(
+                    redisScript, getKeys(key), permitsPerSecond.toString(), capacity.toString(), permits.toString()
+                )
+                acquireResult = results[0] == 1L
+            }
+            logger.info("acquire distributed leaky rateLimiter elapsed time: $elapsedTime")
+            return acquireResult
+        } catch (e: Exception) {
+            e.printStackTrace()
+            throw AcquireLockFailedException("distributed lock acquire failed: $e")
+        }
+    }
+
+    private fun getKeys(key: String): List<String> {
+        return listOf(key, "$key.timestamp")
+    }
+
+
+    companion object {
+        private val logger: Logger = LoggerFactory.getLogger(DistributedLeakyRateLimiter::class.java)
+    }
+}

src/backend/common/common-ratelimiter/src/main/kotlin/com/tencent/bkrepo/common/ratelimiter/algorithm/DistributedSlidingWindowRateLimiter.kt

@@ -0,0 +1,75 @@
+/*
+ * Tencent is pleased to support the open source community by making BK-CI 蓝鲸持续集成平台 available.
+ *
+ * Copyright (C) 2022 THL A29 Limited, a Tencent company.  All rights reserved.
+ *
+ * BK-CI 蓝鲸持续集成平台 is licensed under the MIT license.
+ *
+ * A copy of the MIT License is included in this file.
+ *
+ *
+ * Terms of the MIT License:
+ * ---------------------------------------------------
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of
+ * the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
+ * LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+ * NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+ * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+ * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package com.tencent.bkrepo.common.ratelimiter.algorithm
+
+import com.tencent.bkrepo.common.ratelimiter.exception.AcquireLockFailedException
+import com.tencent.bkrepo.common.ratelimiter.redis.LuaScript
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+import org.springframework.data.redis.core.RedisTemplate
+import org.springframework.data.redis.core.script.DefaultRedisScript
+import java.util.concurrent.TimeUnit
+import kotlin.system.measureTimeMillis
+
+/**
+ * 分布式滑动窗口算法实现
+ */
+class DistributedSlidingWindowRateLimiter(
+    private val key: String,
+    private val limit: Long,
+    private val interval: Long,
+    private val limitUnit: TimeUnit,
+    private val redisTemplate: RedisTemplate<String, String>,
+) : RateLimiter {
+    override fun tryAcquire(permits: Long): Boolean {
+        try {
+            var acquireResult = false
+            val elapsedTime = measureTimeMillis {
+                val redisScript = DefaultRedisScript(LuaScript.slidingWindowRateLimiterScript, List::class.java)
+                val results = redisTemplate.execute(
+                    redisScript, getKeys(key), limit.toString(), (interval * limitUnit.toSeconds(1)).toString(), permits.toString()
+                )
+                acquireResult = results[0] == 1L
+            }
+            logger.info("acquire distributed sliding window rateLimiter elapsed time: $elapsedTime")
+            return acquireResult
+        } catch (e: Exception) {
+            e.printStackTrace()
+            throw AcquireLockFailedException("distributed lock acquire failed: $e")
+        }
+    }
+
+    private fun getKeys(key: String): List<String> {
+        return listOf(key)
+    }
+
+
+    companion object {
+        private val logger: Logger = LoggerFactory.getLogger(DistributedSlidingWindowRateLimiter::class.java)
+    }
+}

src/backend/common/common-ratelimiter/src/main/kotlin/com/tencent/bkrepo/common/ratelimiter/algorithm/DistributedTokenBucketRateLimiter.kt

@@ -45,9 +45,7 @@ class DistributedTokenBucketRateLimiter(
     private val redisTemplate: RedisTemplate<String, String>,
 ) : RateLimiter {
     override fun tryAcquire(permits: Long): Boolean {
-
         try {
-
             var acquireResult = false
             val elapsedTime = measureTimeMillis {
                 val redisScript = DefaultRedisScript(LuaScript.tokenBucketRateLimiterScript, List::class.java)
@@ -62,7 +60,6 @@ class DistributedTokenBucketRateLimiter(
             e.printStackTrace()
             throw AcquireLockFailedException("distributed lock acquire failed: $e")
         }
-
     }
 
     private fun getKeys(key: String): List<String> {

src/backend/common/common-ratelimiter/src/main/kotlin/com/tencent/bkrepo/common/ratelimiter/algorithm/FixedWindowRateLimiter.kt

@@ -31,7 +31,6 @@ import com.google.common.base.Stopwatch
 import com.tencent.bkrepo.common.ratelimiter.constant.TRY_LOCK_TIMEOUT
 import com.tencent.bkrepo.common.ratelimiter.exception.AcquireLockFailedException
 import java.util.concurrent.TimeUnit
-import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.locks.Lock
 import java.util.concurrent.locks.ReentrantLock
 
@@ -44,32 +43,28 @@ class FixedWindowRateLimiter(
     private val stopWatch: Stopwatch = Stopwatch.createStarted()
 ) : RateLimiter {
 
-    private val currentValue: AtomicLong = AtomicLong(0)
-
+    private var currentValue: Long = 0
     private val lock: Lock = ReentrantLock()
 
     override fun tryAcquire(permits: Long): Boolean {
         // TODO 当剩余容量少于permit时，会导致一直获取不到
-        var updateValue = currentValue.incrementAndGet()
-        if (updateValue <= limit) {
-            return true
-        }
         try {
             if (!lock.tryLock(TRY_LOCK_TIMEOUT, TimeUnit.MILLISECONDS)) {
-                throw AcquireLockFailedException("tryLock wait too long: $TRY_LOCK_TIMEOUT ms")
+                throw AcquireLockFailedException("fix window tryLock wait too long: $TRY_LOCK_TIMEOUT ms")
             }
             try {
                 if (stopWatch.elapsed(TimeUnit.MILLISECONDS) > unit.toMillis(1)) {
-                    currentValue.set(0)
+                    currentValue = 0
                     stopWatch.reset()
                 }
-                updateValue = currentValue.addAndGet(permits)
-                return updateValue <= limit
+                if (currentValue + permits > limit) return false
+                currentValue += permits
+                return true
             } finally {
                 lock.unlock()
             }
         } catch (e: InterruptedException) {
-            throw AcquireLockFailedException("tryLock is interrupted by lock timeout: $e")
+            throw AcquireLockFailedException("fix window tryLock is interrupted by lock timeout: $e")
         }
     }
 }

src/backend/common/common-ratelimiter/src/main/kotlin/com/tencent/bkrepo/common/ratelimiter/algorithm/LeakyRateLimiter.kt

@@ -0,0 +1,77 @@
+/*
+ * Tencent is pleased to support the open source community by making BK-CI 蓝鲸持续集成平台 available.
+ *
+ * Copyright (C) 2022 THL A29 Limited, a Tencent company.  All rights reserved.
+ *
+ * BK-CI 蓝鲸持续集成平台 is licensed under the MIT license.
+ *
+ * A copy of the MIT License is included in this file.
+ *
+ *
+ * Terms of the MIT License:
+ * ---------------------------------------------------
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of
+ * the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
+ * LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+ * NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+ * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+ * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package com.tencent.bkrepo.common.ratelimiter.algorithm
+
+import com.tencent.bkrepo.common.ratelimiter.constant.TRY_LOCK_TIMEOUT
+import com.tencent.bkrepo.common.ratelimiter.exception.AcquireLockFailedException
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.locks.Lock
+import java.util.concurrent.locks.ReentrantLock
+
+
+/**
+ * 单机令牌桶算法实现
+ */
+class LeakyRateLimiter(
+    private val permitsPerSecond: Double,
+    private val capacity: Long,
+) : RateLimiter {
+
+    // 计算的起始时间
+    private var lastLeakTime = System.currentTimeMillis()
+    private var water: Long = 0
+    private val lock: Lock = ReentrantLock()
+
+
+    override fun tryAcquire(permits: Long): Boolean {
+        try {
+            if (!lock.tryLock(TRY_LOCK_TIMEOUT, TimeUnit.MILLISECONDS)) {
+                throw AcquireLockFailedException("leaky tryLock wait too long: $TRY_LOCK_TIMEOUT ms")
+            }
+            try {
+                return allow(permits)
+            } finally {
+                lock.unlock()
+            }
+        } catch (e: InterruptedException) {
+            throw AcquireLockFailedException("leaky tryLock is interrupted by lock timeout: $e")
+        }
+    }
+
+    fun allow(permits: Long): Boolean {
+        val now = System.currentTimeMillis()
+        val timeElapsed = now - lastLeakTime
+        water = Math.max(0, (water - timeElapsed * permitsPerSecond / 1000).toLong()) // 漏水
+        lastLeakTime = now
+        if (water + permits <= capacity) {
+            water += permits
+            return true
+        }
+        return false
+    }
+}

src/backend/common/common-ratelimiter/src/main/kotlin/com/tencent/bkrepo/common/ratelimiter/algorithm/SlidingWindowRateLimiter.kt

@@ -0,0 +1,93 @@
+/*
+ * Tencent is pleased to support the open source community by making BK-CI 蓝鲸持续集成平台 available.
+ *
+ * Copyright (C) 2022 THL A29 Limited, a Tencent company.  All rights reserved.
+ *
+ * BK-CI 蓝鲸持续集成平台 is licensed under the MIT license.
+ *
+ * A copy of the MIT License is included in this file.
+ *
+ *
+ * Terms of the MIT License:
+ * ---------------------------------------------------
+ * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+ * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+ * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+ * permit persons to whom the Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all copies or substantial portions of
+ * the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
+ * LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+ * NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+ * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+ * SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ */
+
+package com.tencent.bkrepo.common.ratelimiter.algorithm
+
+import com.tencent.bkrepo.common.ratelimiter.constant.TRY_LOCK_TIMEOUT
+import com.tencent.bkrepo.common.ratelimiter.exception.AcquireLockFailedException
+import java.util.LinkedList
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.locks.Lock
+import java.util.concurrent.locks.ReentrantLock
+
+/**
+ * 单机令牌桶算法实现
+ */
+class SlidingWindowRateLimiter(
+    private val limit: Long,
+    private val interval: Long,
+    private val limitUnit: TimeUnit,
+) : RateLimiter {
+
+    private val queue = LinkedList<Long>()
+    private var counter: Long = 0
+    private var lastUpdate = System.currentTimeMillis()
+    private val lock: Lock = ReentrantLock()
+
+    override fun tryAcquire(permits: Long): Boolean {
+        try {
+            if (!lock.tryLock(TRY_LOCK_TIMEOUT, TimeUnit.MILLISECONDS)) {
+                throw AcquireLockFailedException("sliding window tryLock wait too long: $TRY_LOCK_TIMEOUT ms")
+            }
+            try {
+                return allow(permits)
+            } finally {
+                lock.unlock()
+            }
+        } catch (e: InterruptedException) {
+            throw AcquireLockFailedException("sliding window tryLock is interrupted by lock timeout: $e")
+        }
+    }
+
+    fun allow(permits: Long): Boolean {
+        val now = System.currentTimeMillis()
+        val slots = ((now - lastUpdate) / (interval / limitUnit.toMillis(1))).toInt()
+
+        // 移除过期的时间片
+        while (queue.size >= slots) {
+            val removed = queue.removeFirst()
+            counter -= removed
+        }
+
+        // 添加新的时间片
+        for (i in queue.size until slots) {
+            queue.addLast(0)
+        }
+
+        // 如果当前时间片的请求数量已经达到窗口大小，则拒绝请求
+        if (counter + permits > limit) {
+            return false
+        }
+
+        // 更新当前时间片的请求数量，并增加计数器
+        queue.addLast(queue.removeLast() + permits)
+        counter += permits
+        lastUpdate = now
+        return true
+    }
+
+}