高并发访问时,缓存、限流、降级往往是系统的利剑,在互联网蓬勃发展的时期,经常会面临因用户暴涨导致的请求不可用的情况,甚至引发连锁反映导致整个系统崩溃。这个时候常见的解决方案之一就是限流了,当请求达到一定的并发数或速率,就进行等待、排队、降级、拒绝服务等…
令牌桶算法
令牌桶算法的原理是系统会以一个恒定的速度往桶里放入令牌,而如果请求需要被处理,则需要先从桶里获取一个令牌,当桶里没有令牌可取时,则拒绝服务。 当桶满时,新添加的令牌被丢弃或拒绝。
在Google Guava 中提供了一个 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 - 拒絕...