详解几种常见限流算法及限流器设计

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class TokenBucketRateLimiter {
    private final int bucketCapacity; // 令牌桶的容量
    private final int refillRate; // 令牌生成速率（每秒生成的令牌数）
    private AtomicInteger tokens; // 当前令牌数

    public TokenBucketRateLimiter(int bucketCapacity, int refillRate) {
        this.bucketCapacity = bucketCapacity;
        this.refillRate = refillRate;
        this.tokens = new AtomicInteger(0);
        startRefilling(); // 启动定期生成令牌的线程
    }

    // 尝试获取一个令牌
    public boolean tryAcquire() {
        int currentTokens = tokens.get();
        if (currentTokens > 0) {
            if (tokens.compareAndSet(currentTokens, currentTokens - 1)) {
                System.out.println("请求获得令牌，允许通过");
                return true;
            }
        }
        System.out.println("请求被拒绝：无可用令牌");
        return false;
    }

    // 定期为令牌桶添加令牌
    private void startRefilling() {
        ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
        executor.scheduleAtFixedRate(() -> {
            int currentTokens = tokens.get();
            if (currentTokens < bucketCapacity) {
                int newTokens = Math.min(bucketCapacity, currentTokens + refillRate);
                tokens.set(newTokens);
                System.out.println("新增令牌，当前令牌数：" + newTokens);
            }
        }, 0, 1, TimeUnit.SECONDS); // 每秒填充令牌
    }

    public static void main(String[] args) {
        TokenBucketRateLimiter rateLimiter = new TokenBucketRateLimiter(10, 2); // 容量为10，每秒生成2个令牌

        // 模拟10个请求
        for (int i = 0; i < 10; i++) {
            boolean allowed = rateLimiter.tryAcquire();
            if (!allowed) {
                System.out.println("请求 " + i + " 被拒绝");
            }
            try {
                Thread.sleep(200); // 模拟请求间隔
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

package ratelimit

import (
  "fmt"
  "sync"
  "time"
 )

 type RateLimiter struct {
  rate   int64 // 令牌放入速度
  max    int64 // 令牌最大数量
  last   int64 // 上一次请求发生时间
  amount int64 // 令牌数量
  lock   sync.Mutex // 由于读写冲突，需要加锁
 }

    // 获得当前时间
 func cur() int64 {
  return time.Now().Unix()
 }

 func New(rate int64, max int64) *RateLimiter {
  // TODO: 检查一下rate和max是否合法
  return &RateLimiter{
   rate:   rate,
   max:    max,
   last:   cur(),
   amount: max,
  }
 }

 func (rl *RateLimiter) Pass() bool {
  rl.lock.Lock()
  defer rl.lock.Unlock()

        // 距离上一次请求过去的时间
  passed := cur() - rl.last
  fmt.Println("passed is: ", passed)

        // 计算在这段时间里 令牌数量可以增加多少
  amount := rl.amount + passed*rl.rate

        // 如果令牌数量超过上限；我们就不继续放入那么多令牌了
  if amount > rl.max {
   amount = rl.max
  }

        // 如果令牌数量仍然小于0，则说明请求应该拒绝
  if amount <= 0 {
   return false
  }

        // 请求被放行则令牌数-1
  amount--
  rl.amount = amount
        // 更新上次请求时间
  rl.last = cur()

  return true
 }

import com.google.common.util.concurrent.RateLimiter;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RateLimiterExample {

    public static void main(String[] args) {
        // 创建一个RateLimiter，每秒允许处理5个请求
        RateLimiter rateLimiter = RateLimiter.create(5.0);

        // 创建一个固定大小的线程池
        ExecutorService executorService = Executors.newFixedThreadPool(10);

        // 模拟处理20个请求
        for (int i = 0; i < 20; i++) {
            final int requestNumber = i;
            executorService.submit(() -> handleRequest(requestNumber, rateLimiter));
        }

        // 关闭线程池
        executorService.shutdown();
    }

    private static void handleRequest(int requestNumber, RateLimiter rateLimiter) {
        // 获取一个令牌，可能会阻塞直到令牌可用
        rateLimiter.acquire();

        // 模拟处理请求的逻辑
        System.out.println("Handling request " + requestNumber + " at " + System.currentTimeMillis());
    }
}