探索常见的几种限流策略和实现

xkl135 2018-07-29

展开全文

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

限流算法

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

令牌桶算法

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

如图所示

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

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

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

示例代码

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 毫秒左右

当前时间 - 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）先进先出，而非公平模式则是凭本事抢资源，想先进先出？不存在的。默认是非公平的

示例代码

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 个任务，超出进入缓冲区排队等待，当缓冲区满了后则拒绝接收新的请求...

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

示例代码

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);
}
}

运行日志

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 - 拒絕...