限流探索

原创
Service
文章目录
  1. 1. 限流算法
    1. 1.1. 令牌桶算法
    2. 1.2. 漏桶算法
    3. 1.3. 计数器限流
  2. 2. 总结
  3. 3. 说点什么

高并发访问时，缓存、限流、降级往往是系统的利剑，在互联网蓬勃发展的时期，经常会面临因用户暴涨导致的请求不可用的情况，甚至引发连锁反映导致整个系统崩溃。这个时候常见的解决方案之一就是限流了，当请求达到一定的并发数或速率，就进行等待、排队、降级、拒绝服务等…

限流算法

常见的算法有漏桶(leaky bucket)、令牌桶(Token Bucket)、计数器，本章通过最简单的代码和最直白的文字描述三种的实现方式（基于本地而不是分布式）…

令牌桶算法

令牌桶算法的原理是系统会以一个恒定的速度往桶里放入令牌，而如果请求需要被处理，则需要先从桶里获取一个令牌，当桶里没有令牌可取时，则拒绝服务。 当桶满时，新添加的令牌被丢弃或拒绝。

如图所示

令牌桶算法

在Google Guava 中提供了一个 RateLimiter 工具类，就是基于令牌桶算法实现平滑突发的限流策略，令牌桶的好处是可以方便的改变速度. 一旦需要提高速率,则按需提高放入桶中的令牌的速率. 一般会定时(比如1000毫秒)往桶中增加一定数量的令牌, 有些变种算法则可以实时的计算应该增加的令牌的数量

在示例代码中为每秒中产生 2 个令牌，意味着每500毫秒会产生一个令牌。

limiter.acquire(num) 表示消费多少个令牌。当桶中有足够的令牌时，则直接返回0，否则阻塞，直到有可用的令牌数才返回，返回的值为阻塞的时间。

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
package com.battcn.limiting;

import com.google.common.util.concurrent.RateLimiter;
import org.junit.Test;

import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author Levin
 * @since 2018/7/24 0024
 */
public class RateLimiterTest {
    /**
     * 令牌桶算法
     * 每秒生成 2 個令牌
     */
    private static final RateLimiter limiter = RateLimiter.create(2);

    private void rateLimiter() {
        // 默认就是 1
        final double acquire = limiter.acquire(1);
        System.out.println("当前时间 - " + LocalDateTime.now() + " - " + Thread.currentThread().getName() + " - 阻塞 - " + acquire + " 通过...");
    }

    @Test
    public void testDemo1() throws InterruptedException {
        final ExecutorService service = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 5; i++) {
            service.execute(this::rateLimiter);
        }
        TimeUnit.SECONDS.sleep(5);
    }
}

运行日志

通过日志可以很直观的看出每个令牌产生时间间隔大约在 500 毫秒左右

1
2
3
4
5
当前时间 - 2018-07-24T15:10:25.091 - pool-1-thread-2 - 阻塞 - 0.0 通过...
当前时间 - 2018-07-24T15:10:25.531 - pool-1-thread-5 - 阻塞 - 0.453466 通过...
当前时间 - 2018-07-24T15:10:26.031 - pool-1-thread-3 - 阻塞 - 0.953454 通过...
当前时间 - 2018-07-24T15:10:26.531 - pool-1-thread-4 - 阻塞 - 1.453354 通过...
当前时间 - 2018-07-24T15:10:27.031 - pool-1-thread-1 - 阻塞 - 1.952553 通过...

漏桶算法

其主要目的是控制数据注入到网络的速率，平滑网络上的突发流量，数据可以以任意速度流入到漏桶中。 漏桶算法提供了一种机制，通过它，突发流量可以被整形以便为网络提供一个稳定的流量。 漏桶可以看作是一个带有常量服务时间的单服务器队列，如果漏桶为空，则不需要流出水滴，如果漏桶（包缓存）溢出，那么水滴会被溢出丢弃。

如图所示

漏桶算法

漏铜算法可以通过 信号量（Semaphore） 的方式实现，很好的达到消峰的目的，如果下文中的代码，队列中任务存活个数就如同是水桶最多能盛装的水量，当超出这个阀值就会丢弃任务….

Semaphore 是 JDK1.5 提供用于限制获取某种资源的线程数量，拥有有 公平、非公平 两种模式。公平则是顺序获取信号，遵循（FIFO）先进先出，而非公平模式则是凭本事抢资源，想先进先出？不存在的。默认是非公平的

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
package com.battcn.limiting;

import org.junit.Test;

import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * @author Levin
 * @since 2018/7/23 0023
 */
public class SemaphoreLimiterTest {

    /**
     * 计数器限流算法（允许将任务放入到缓冲队列）
     * 信号量，用来达到削峰的目的
     */
    private static final Semaphore semaphore = new Semaphore(3);


    private void semaphoreLimiter() {
        // 队列中允许存活的任务个数不能超过 5 个
        if (semaphore.getQueueLength() > 5) {
            System.out.println(LocalDateTime.now() + " - " + Thread.currentThread().getName() + " - 拒絕...");
        } else {
            try {
                semaphore.acquire();
                System.out.println(LocalDateTime.now() + " - " + Thread.currentThread().getName() + " - 通过...");
                //处理核心逻辑
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release();
            }
        }
    }

    @Test
    public void testSemaphore() throws InterruptedException {
        final ExecutorService service = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            service.execute(this::semaphoreLimiter);
        }
        TimeUnit.SECONDS.sleep(5);
    }
}

运行日志

一共 10 个线程同时请求，初始信号量为3，表示最多可以同时处理 3 个任务，超出进入缓冲区排队等待，当缓冲区满了后则拒绝接收新的请求…

1
2
3
4
5
6
7
8
9
10
2018-07-24T15:17:00.236 - pool-1-thread-3 - 通过...
2018-07-24T15:17:00.237 - pool-1-thread-2 - 通过...
2018-07-24T15:17:00.236 - pool-1-thread-10 - 拒絕...
2018-07-24T15:17:00.237 - pool-1-thread-1 - 通过...
2018-07-24T15:17:02.237 - pool-1-thread-4 - 通过...
2018-07-24T15:17:02.237 - pool-1-thread-5 - 通过...
2018-07-24T15:17:02.237 - pool-1-thread-6 - 通过...
2018-07-24T15:17:04.238 - pool-1-thread-9 - 通过...
2018-07-24T15:17:04.238 - pool-1-thread-7 - 通过...
2018-07-24T15:17:04.238 - pool-1-thread-8 - 通过...

计数器限流

计数器限流算法是比较常用一种的限流方案也是最为粗暴直接的，主要用来限制总并发数，比如数据库连接池大小、线程池大小、接口访问并发数等都是使用计数器算法…

使用 AomicInteger 来进行统计当前正在并发执行的次数，如果超过域值就直接拒绝请求，提示系统繁忙

AtomicInteger 是 JDK1.5 提供的拥有原子特性的计数功能，都知道在多线程环境下 num++ 是非原子的，但是有了 AtomicInteger 后这个问题可以非常简单的解决，它就像是 redis incr

示例代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
package com.battcn.limiting;

import org.junit.Test;

import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Levin
 * @since 2018/7/24 0024
 */
public class AtomicLimiterTest {

    /**
     * 计数器限流算法（比较暴力/超出直接拒绝）
     * Atomic，限制总数
     */
    private static final AtomicInteger atomic = new AtomicInteger(0);

    private void atomicLimiter() {
        // 最大支持 3 個
        if (atomic.get() >= 3) {
            System.out.println(LocalDateTime.now() + " - " + Thread.currentThread().getName() + " - " + "拒絕...");
        } else {
            try {
                atomic.incrementAndGet();
                //处理核心逻辑
                System.out.println(LocalDateTime.now() + " - " + Thread.currentThread().getName() + " - " + "通过...");
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                atomic.decrementAndGet();
            }
        }
    }

    @Test
    public void testAtomic() throws InterruptedException {
        final ExecutorService service = Executors.newFixedThreadPool(5);
        for (int i = 0; i < 5; i++) {
            service.execute(this::atomicLimiter);
        }
        TimeUnit.SECONDS.sleep(5);
    }
}

运行日志

1
2
3
4
5
2018-07-24T15:31:40.270 - pool-1-thread-2 - 通过...
2018-07-24T15:31:40.271 - pool-1-thread-4 - 拒絕...
2018-07-24T15:31:40.271 - pool-1-thread-1 - 通过...
2018-07-24T15:31:40.271 - pool-1-thread-3 - 通过...
2018-07-24T15:31:40.271 - pool-1-thread-5 - 拒絕...

总结

本篇文章简短的介绍了几种限流方案，虽然是针对本地限流的方案而不是分布式的，但学习本来就是循环渐进的，由浅入深….

说点什么

  • 个人QQ：1837307557
  • battcn开源群（适合新手）：391619659
  • 微信公众号（欢迎调戏）：battcn

公众号

个人博客：http://blog.battcn.com/

全文代码：https://github.com/battcn/spring-boot2-learning/tree/master/chapter26

分享到