【开发经验】java代码中实现限流

限流目的

限流的目的是防止恶意请求流量、恶意攻击、或者防止流量超过系统峰值。

流量达到峰值时，会有一个熔断策略，常见的熔断策略：

直接拒绝请求，跳转到一个“服务器出小差”页面
排队等待，比如火车票等待。
服务降级，返回最基本的用户可接受的数值。
- 到达峰值的时候进行排队。
- 定位经纬度定位，精确定位到达峰值，返回大概定位，比如百度地图，有时候可以定位到自己的详细位置，有时候显示自己所在的城市。

限流实现的方式有很多种，比如nginx限流，网关限流，或者在代码中限流。具体如何限流，也要根据不同的场景进行选择。以下介绍下如何在代码中进行限流。

限流方式

漏桶

漏桶是有一个进水口，一个出水口。桶本身具有一个恒定的速率往下漏水，而上方时快时慢的会有水进入桶内。进水速度可以大于出水速度，但是多余的水会积在桶中，一旦水满，上方的水就无法加入。

特点：漏水速度固定。

令牌桶

令牌桶是一个存放固定容量令牌（token）的桶，按照固定速率往桶里添加令牌。当请求到达当时，从桶中拿令牌，有令牌即可进行请求。

限流实现

单机限流

1.Semaphore限流

Semaphore是jdk自带的一种并发限流方式。通过new Semaphore(2);可设置并发数为2。切记获取资源后，要进行资源释放。

public static void main(String[] args) {//并发为2个    Semaphore semaphore = new Semaphore(2);Random random = new Random();for(int i = 0 ; i<5;i++){new Thread(()->{try {semaphore.acquire();System.out.println(Thread.currentThread().getName()+"获取权限");Thread.sleep(random.nextInt(2000)); // 仿造处理时间} catch (InterruptedException e) {e.printStackTrace();}finally {semaphore.release(); //注意一定要finally中进行释放资源System.out.println(Thread.currentThread().getName()+"释放");}}).start();}}

Semaphore.acquire/release方法要配对使用，使用acquire申请资源，使用release释放资源。Semaphore即使在没有申请到资源的情况下，还是可以通过release释放资源，这里就要自己通过代码进行合适的处理。和锁不同的是，锁必须获得锁才能释放锁，这里并没有这种限制。
Semaphore.release方法尽可能的放到finally中避免资源无法归还。
如果Semaphore的配额为1，那么创建的实例相当与一个互斥锁，由于可以在没有申请资源的情况调用release释放资源，所以，这里允许一个线程释放另一个线程的锁。

2.RateLimiter限流

Guava工程包含了若干被Google的 Java项目广泛依赖的核心库，例如：集合 [collections] 、缓存 [caching] 、原生类型支持 [primitives support] 、并发库 [concurrency libraries] 、通用注解 [common annotations] 、字符串处理 [string processing] 、I/O 等等。所有这些工具每天都在被Google的工程师应用在产品服务中。更多信息。

使用Guava的RateLimiter类通过令牌桶进行限流。

引入jar

       <dependency><groupId>com.google.guava</groupId><artifactId>guava</artifactId><version>28.1-jre</version></dependency>

1普通令牌桶

 public static void main(String[] args) {RateLimiter r = RateLimiter.create(5);for(;;){System.out.println("get 1 tokens: " + r.acquire() + "s");}/*** output: 基本上都是0.2s执行一次，符合一秒发放5个令牌的设定。* get 1 tokens: 0.0s* get 1 tokens: 0.182014s* get 1 tokens: 0.188464s* get 1 tokens: 0.198072s* get 1 tokens: 0.196048s* get 1 tokens: 0.197538s* get 1 tokens: 0.196049s*/}

通过RateLimiter.create(5);方法声明每秒创建5个令牌。
acquire方法进行获取令牌，并且返回获取时间。

令牌积累情况

 public static void main(String[] args) throws InterruptedException {RateLimiter r = RateLimiter.create(2);for(;;){System.out.println("get 1 tokens: " + r.acquire() + "s");System.out.println("get 1 tokens: " + r.acquire() + "s");Thread.sleep(2000l);System.out.println("get 1 tokens: " + r.acquire() + "s");System.out.println("get 1 tokens: " + r.acquire() + "s");System.out.println("get 1 tokens: " + r.acquire() + "s");System.out.println("get 1 tokens: " + r.acquire() + "s");System.out.println("get 1 tokens: " + r.acquire() + "s");}/*** get 1 tokens: 0.0s* get 1 tokens: 0.49874s*  休眠2 秒* get 1 tokens: 0.0s* get 1 tokens: 0.0s  // 积累了两个* get 1 tokens: 0.0s  // 预支1个,预支是什么？下面说* get 1 tokens: 0.499827s* get 1 tokens: 0.499159s* get 1 tokens: 0.49749s* get 1 tokens: 0.499572s*/}

在没有足够令牌发放时，采用滞后处理的方式，也就是前一个请求获取令牌所需等待的时间由下一次请求来承受，也就是代替前一个请求进行等待。

    public static void main(String[] args) throws InterruptedException {RateLimiter r = RateLimiter.create(2);System.out.println("get 1 tokens: " + r.acquire(4) + "s");System.out.println("get 1 tokens: " + r.acquire() + "s");System.out.println("get 1 tokens: " + r.acquire() + "s");System.out.println("get 1 tokens: " + r.acquire() + "s");/**get 1 tokens: 0.0s        第一次获取4个get 1 tokens: 1.998205s  ，第二次获取要为第一次买单，所以等待两秒，如果第一次获取6个，则就是等待3秒。get 1 tokens: 0.496947sget 1 tokens: 0.499579s*/}

2.预热令牌桶

RateLimiter.create(2, 3, TimeUnit.SECONDS);说明需要3秒的预热时间，开始的时候是达不到每秒中2个令牌，在3秒内创建令牌速度逐渐加快，最后达到每秒钟2个令牌。
预热令牌桶场景：可能有缓存失效的时候，这段时间需要和数据库交互，进行一个缓存的预热。

public static void main(String[] args) {RateLimiter r = RateLimiter.create(2, 3, TimeUnit.SECONDS);for (;;){System.out.println("get 1 tokens: " + r.acquire(1) + "s");System.out.println("get 1 tokens: " + r.acquire(1) + "s");System.out.println("get 1 tokens: " + r.acquire(1) + "s");System.out.println("get 1 tokens: " + r.acquire(1) + "s");System.out.println("end");/*** output:* get 1 tokens: 0.0s* get 1 tokens: 1.329289s* get 1 tokens: 0.994375s* get 1 tokens: 0.662888s  上边三次获取的时间相加正好为3秒* end* get 1 tokens: 0.49764s  正常速率0.5秒一个令牌* get 1 tokens: 0.497828s* get 1 tokens: 0.49449s* get 1 tokens: 0.497522s*/}
}

参考资料：超详细的Guava RateLimiter限流原理解析

集群限流

redis限流

在正式的开发过程中，往往是集群化部署的，那么上面的单机限流就力不从心了。可以采用redis进行集群限流。redission框架通过lua脚本实现的集群限流。
redission通过getRateLimiter方法进行限流，使用方式和guava的限流类似。

  public static void main(String[] args) {Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379");RedissonClient client = Redisson.create(config);RRateLimiter rateLimiter = client.getRateLimiter("order"); RateType.OVERALL,    //所有客户端加总限流，就是集群下所有的流量 RateType.PER_CLIENT; //每个客户端单独计算流量，每台机器的流量rateLimiter.trySetRate(RateType.OVERALL, 2, 1, RateIntervalUnit.SECONDS);ExecutorService executorService = Executors.newFixedThreadPool(10);for (int i = 0; i < 10; i++) {executorService.submit(() -> {try {long start = System.currentTimeMillis();rateLimiter.acquire();System.out.println("线程" + Thread.currentThread().getName() + "进入数据区：" + (start-System.currentTimeMillis()));} catch (Exception e) {e.printStackTrace();}});}/*** 线程pool-1-thread-4进入数据区：-10* 线程pool-1-thread-5进入数据区：-13* 线程pool-1-thread-7进入数据区：-1010* 线程pool-1-thread-10进入数据区：-1013* 线程pool-1-thread-1进入数据区：-2017* 线程pool-1-thread-9进入数据区：-2016* 线程pool-1-thread-8进入数据区：-3021* 线程pool-1-thread-3进入数据区：-3022* 线程pool-1-thread-2进入数据区：-4028* 线程pool-1-thread-6进入数据区：-4028*/}

Sentinel限流

Sentinel 以流量为切入点，从流量控制、熔断降级、系统负载保护等多个维度保护服务的稳定性。
Sentinel 具有以下特征:

丰富的应用场景：Sentinel 承接了阿里巴巴近 10 年的双十一大促流量的核心场景，例如秒杀（即突发流量控制在系统容量可以承受的范围）、消息削峰填谷、集群流量控制、实时熔断下游不可用应用等。
完备的实时监控：Sentinel 同时提供实时的监控功能。您可以在控制台中看到接入应用的单台机器秒级数据，甚至 500 台以下规模的集群的汇总运行情况。
广泛的开源生态：Sentinel 提供开箱即用的与其它开源框架/库的整合模块，例如与 Spring Cloud、Dubbo、gRPC 的整合。您只需要引入相应的依赖并进行简单的配置即可快速地接入 Sentinel。
完善的 SPI 扩展点：Sentinel 提供简单易用、完善的 SPI 扩展接口。您可以通过实现扩展接口来快速地定制逻辑。例如定制规则管理、适配动态数据源等。

Sentinel官方文档