分布式限流

2020-06-11 本文已影响0人 __y

1.系统优化应对流量激增的手段

当我们流量激增的时候，为了保持系统的对外高可用，即不能让系统崩溃状态。常用的思路是以下方面：

限流：对应用入口流量做控制，瞬时流量向后迁移，对下游请求流量做自适应限流，根据接口响应时间动态调整流量。
延迟排队：如果请求量大，按业务线优先级排队，优先保障线上渠道实时的请求。（使用MQ削峰）
路由：这个是因为业务的特殊性，所有的请求都依赖下游第三方的服务，可以将多家下游服务供应商做个动态路由表，将请求优先路由给接口成功率高、耗时低的服务供应商；
备份：这基本是所有分布式组件都会做的，能做多机的不做单机，例如：Redis 做三主三备(集群)、MySQL分库分表、MQ 与 Redis 互为备份等等；
降级：这个是最后的迫不得已的措施，如果遇到全线崩溃，使用降级手段保障系统核心功能可用，或让模块达到最小可用。
日志：完整的监控和链路日志，日志功能很多，也分很多种，一方面是方便排查问题，另一方面可用来做任务重试、任务回滚、数据恢复、状态持久化等。

2.限流

限流，顾名思义，就是限制流量，一般分为限制入口流量和限制出口流量，入口流量是人家来请求我的系统，我在入口处加了一道阀门，出口流量是我调外部系统，我在出口加一道阀门。简而言之，就是有一道门，就像你过安检一样，每次只能通过若干的人数。

3.限流的实现方法

3.1 单机的限流实现方法

如果是单机，可以通过Semphore 限制统一时间请求接口的量，也可以用 Google Guava 包提供的限流包
比如:我们现在有5台机器，但是有8个工人;这个时候工人和机器是不对等的。那怎么办呢，那肯定一批一批上啊，先上5个人，然后再让其他3个工人进行操作。下面用代码进行演示

package threadTest;

import java.util.concurrent.Semaphore;

public class Main5 {
    public static void main(String[] args) {
        int n = 8;
        Semaphore semaphore = new Semaphore(5);
        for(int i=0; i<n; i++) {
            new Test(i,semaphore).start();
        }
    }
    static class Test extends Thread{
        private int num;
        private Semaphore semaphore;
        Test(int num, Semaphore semaphore) {
            this.num = num;
            this.semaphore = semaphore;
        }

        @Override
        public void run() {
            try {
                semaphore.acquire();
                System.out.println("工人" + this.num + "占用一个机器在生产......");
                semaphore.release();
                System.out.println("工人" + this.num + "休息去了(释放)......");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
    }
}

image.png

3.2 分布式限流

如果是分布式环境，可以使用 Redis 实现，也有阿里 Sentinal 或 Spring Cloud Gateway 可以实现限流。
其思想和单机是一样的，也是控制资源的访问频率，一般主流的设计思想有二种：
漏洞算法

image.png

把请求比作水，在请求入口和响应请求的服务之间加一个漏桶，桶中的水以恒定的速度流出，这样保证了服务接收到的流量速度是稳定的，如果桶里的水满了，再进来的水就直接溢出（请求直接拒绝）
漏桶是网络环境中流量整形（Traffic Shaping）或速率限制（Rate Limiting）时经常使用的一种算法，它的主要目的是控制数据进入到网络的速率，平滑网络上的突发流量。
令牌桶算法

image.png
令牌桶算法有点类似于生产者消费者模式，专门有一个生产者往令牌桶中以恒定速率放入令牌，而请求处理器(消费者)在处理请求时必须先从桶中获得令牌，如果没有拿到令牌，有二种策略：一种是直接返回拒绝请求，一种是等待一段时间，再次尝试获取令牌
令牌桶算法用来控制发送到网络上的数据的数目，并允许突发数据的发送

3.3 redis实现分布式限流

废话不多说，直接上lua脚本

local tokens_key = KEYS[1]
local timestamp_key = KEYS[2]

local rate = tonumber(ARGV[1])
local capacity = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])

local fill_time = capacity/rate
local ttl = math.floor(fill_time*2)


local last_tokens = tonumber(redis.call("get", tokens_key)) 

if last_tokens == nil then
  last_tokens = capacity
end

local last_refreshed = tonumber(redis.call("get", timestamp_key)) 

if last_refreshed == nil then
  last_refreshed = 0
end

local delta = math.max(0, now-last_refreshed)  
local filled_tokens = math.min(capacity, last_tokens+(delta*rate))

local allowed = filled_tokens >= requested      
local new_tokens = filled_tokens
local allowed_num = 0
if allowed then
  new_tokens = filled_tokens - requested
  allowed_num = 1
end       

redis.call("setex", tokens_key, ttl, new_tokens)
redis.call("setex", timestamp_key, ttl, now)

return { allowed_num, new_tokens }

PS：这里可能会问，lua 脚本的执行会不会有性能上的损耗，比较redis是单线程的？
redis 使用 epoll 实现I/O多路复用的事件驱动模型，对于每一个读取和写入操作都尽量要快速
可以使用以下方式进行压测：
1.通过script load 命令加载redis lua脚本，得到sha1 之后直接运行

// 1. 在redis服务端load 脚本 拿到sha
redis-cli script load "$(cat ratelimit.lua)"
//sha1: ebbcd2ed99990afaca6d2ba61a0f2d5bdd907e59
// 2. 通过脚本 sha1 值运行脚本
redis-cli evalsha ebbcd2ed99990afaca6d2ba61a0f2d5bdd907e59 2 remain.${0}.tokens last_fill_time 0.2 12 `gdate +%s%3N` 1

2.通过redis客户端的压测工具

redis-benchmark -n 100000 evalsha ebbcd2ed99990afaca6d2ba61a0f2d5bdd907e59 2 remain.${1}.tokens last_fill_time 0.2 12 `gdate +%s%3N` 1

image.png
99.9%都在 2ms以内完成，每秒钟执行4万5千多次，因此损耗可以接受。
SpringBoot实现

3.1 将Lua脚本放在resource

image.png

3.2 工程加载脚本

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.List;

@Configuration
@Slf4j
public class LuaConfiguration {

    public static final String RATE_LIMIT_SCRIPT_LOCATION = "scripts/redis_limit.lua";

    @Bean(name = "rateLimitRedisScript")
    public DefaultRedisScript<List> redisScript(LettuceConnectionFactory lettuceConnectionFactory) throws UnsupportedEncodingException {
        DefaultRedisScript<List> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource(RATE_LIMIT_SCRIPT_LOCATION)));
        String rateLimitSha1 = redisScript.getSha1();
        log.info("分布式限流lua脚本 sha1 :{}",rateLimitSha1);
        log.info("lua脚本 script:{}",redisScript.getScriptAsString());

        List<Boolean> luaScriptsExists;

        RedisConnection redisConnection = lettuceConnectionFactory.getConnection();
        if ((luaScriptsExists = redisConnection.scriptExists(redisScript.getSha1())) != null && luaScriptsExists.size() > 0) {
            log.info("redis 已经存在 redis lua脚本 sha1 ：{}",rateLimitSha1);
        } else {
            String scriptLuaSha1 = redisConnection.scriptLoad(redisScript.getScriptAsString().getBytes(StandardCharsets.UTF_8));
            log.info("加载 redis lua 成功 sha1 :{}",scriptLuaSha1);
        }
        return redisScript;
    }

}

3.3 定义加载令牌的工具类

@Component
@Slf4j
public class RateLimiter2 {
    @Autowired
    private RedisLockUtil redisLockUtil;

    @Autowired
    @Qualifier("rateLimitRedisScript")
    private DefaultRedisScript<List> rateLimitRedisScript;
    /**
     * redis集群下；用{1}remain_tokens
     */
    private static final String REDIS_KEY_REMAIN_TOKENS = "remain_tokens";
    private static final String REDIS_KEY_LAST_FILL_TIME = "last_fill_time";

    public boolean achieveDistributeToken(String keySuffix,int tokenCapacity, float tokenGenerateRate,int achiveTokenPer) {
        String remainTokenKey = REDIS_KEY_REMAIN_TOKENS + "_" + keySuffix;
        String lastFillTimeKey = REDIS_KEY_LAST_FILL_TIME + "_" + keySuffix;
        List<String> keys = Arrays.asList(remainTokenKey,lastFillTimeKey);
        Jedis jedis = redisLockUtil.getJedis();
        String now = String.valueOf(System.currentTimeMillis()/1000);
        List<String> args = Arrays.asList(String.valueOf(tokenGenerateRate),String.valueOf(tokenCapacity),now,String.valueOf(achiveTokenPer));
        List<String> result = (List<String>)jedis.eval(rateLimitRedisScript.getScriptAsString(),keys,args);

        if (result != null && result.size() > 0) {
            log.info(">>> 获取分布式令牌是否成功{},接口:{},剩余令牌数量:{}",result.get(0),keySuffix,result.get(1));
            return true;
        }
        return false;
    }
}