基于 Redis 实现高级限流器及其在队列任务处理中的应用

use Illuminate\Support\Facades\Redis;

// 浏览文章
public function show($id)
{
    // 定义一个限定并发访问频率的限流器，最多支持 100 个并发请求
    Redis::funnel("posts.${id}.show.concurrency")
        ->limit(100)
        ->then(function () use ($id) {
            // 正常访问
            $post = $this->postRepo->getById($id);
            event(new PostViewed($post));
            return "Show Post #{$post->id}, Views: {$post->views}";
        }, function () {
            // 触发并发访问上限
            abort(429, 'Too Many Requests');
        });
}

// 定义一个单位时间内限定请求上限的限流器，每 10 秒最多支持 100 个请求
Redis::throttle("posts.${id}.show.concurrency")
    ->allow(100)->every(10)
    ->then(function () use ($id) {
        // 正常访问
        $post = $this->postRepo->getById($id);
        event(new PostViewed($post));
        return "Show Post #{$post->id}, Views: {$post->views}";
    }, function () {
        // 触发并发访问上限
        abort(429, 'Too Many Requests');
    });

public function then(callable $callback, callable $failure = null)
{
    try {
        return (new ConcurrencyLimiter(
            $this->connection, $this->name, $this->maxLocks, $this->releaseAfter
        ))->block($this->timeout, $callback);
    } catch (LimiterTimeoutException $e) {
        if ($failure) {
            return $failure($e);
        }

        throw $e;
    }
}

public function block($timeout, $callback = null)
{
    $starting = time();

    $id = Str::random(20);

    while (! $slot = $this->acquire($id)) {
        if (time() - $timeout >= $starting) {
            throw new LimiterTimeoutException;
        }

        usleep(250 * 1000);
    }

    if (is_callable($callback)) {
        try {
            return tap($callback(), function () use ($slot, $id) {
                $this->release($slot, $id);
            });
        } catch (Exception $exception) {
            $this->release($slot, $id);

            throw $exception;
        }
    }

    return true;
}

protected function acquire($id)
{
    $slots = array_map(function ($i) {
        return $this->name.$i;
    }, range(1, $this->maxLocks));

    return $this->redis->eval(...array_merge(
        [$this->lockScript(), count($slots)],
        array_merge($slots, [$this->name, $this->releaseAfter, $id])
    ));
}

protected function lockScript()
{
    return <<<'LUA'
for index, value in pairs(redis.call('mget', unpack(KEYS))) do
    if not value then
        redis.call('set', KEYS[index], ARGV[3], "EX", ARGV[2])
        return ARGV[1]..index
    end
end
LUA;
}

public function block($timeout, $callback = null)
{
    $starting = time();

    // 获取锁操作
    while (! $this->acquire()) {
        // 超时抛出异常
        if (time() - $timeout >= $starting) {
            throw new LimiterTimeoutException;
        }

        // 获取失败，则阻塞 750ms 后重试
        usleep(750 * 1000);
    }

    // 获取锁成功（还未触发上限），则执行回调函数
    if (is_callable($callback)) {
        return $callback();
    }

    return true;
}

public function acquire()
{
    // 通过 Redis Lua 脚本设置锁，然后从返回值读取信息
    $results = $this->redis->eval(
        $this->luaScript(), 1, $this->name, microtime(true), time(), $this->decay, $this->maxLocks
    );

    // 时间窗口过期时间点（当前时间 + 传入的时间窗口大小）
    $this->decaysAt = $results[1];

    // 剩余支持的请求槽位（传入的请求上限 - 已处理请求数）
    $this->remaining = max(0, $results[2]);

    // 是否获取锁成功（基于是否还有剩余请求槽位判断）
    return (bool) $results[0];
}

protected function luaScript()
{
    return <<<'LUA'
local function reset()
    redis.call('HMSET', KEYS[1], 'start', ARGV[2], 'end', ARGV[2] + ARGV[3], 'count', 1)
    return redis.call('EXPIRE', KEYS[1], ARGV[3] * 2)
end

-- 第一次请求会初始化一个 Hash 结构作为限流器，键名是外部传入的名称，键值是包含起始时间、结束时间和请求统计数的对象
-- 返回值的第一个对应的是是否获取锁成功，即是否可以继续请求，第二个是有效期结束时间点，第三个是剩余的请求槽位数
if redis.call('EXISTS', KEYS[1]) == 0 then
    return {reset(), ARGV[2] + ARGV[3], ARGV[4] - 1}
end

-- 如果限流器已存在，并且还处于有效期对应的时间窗口内，则对请求统计数做自增操作
-- 这里，我们会限定其值不能超过请求上限，否则获取锁失败，有效期结束时间点不便，剩余槽位数=请求上限-当前请求统计数
if ARGV[1] >= redis.call('HGET', KEYS[1], 'start') and ARGV[1] <= redis.call('HGET', KEYS[1], 'end') then
    return {
        tonumber(redis.call('HINCRBY', KEYS[1], 'count', 1)) <= tonumber(ARGV[4]),
        redis.call('HGET', KEYS[1], 'end'),
        ARGV[4] - redis.call('HGET', KEYS[1], 'count')
    }
end

-- 如果限流器已过期，则和第一个请求一样，重置这个限流器，重新开始统计
return {reset(), ARGV[2] + ARGV[3], ARGV[4] - 1}
LUA;
    }
}

public function handle()
{
    Redis::funnel('post.views.increment')
        ->limit(60)
        ->then(function () {
            // 队列任务正常处理逻辑
            if ($this->post->increment('views')) {
                Redis::zincrby('popular_posts', 1, $this->post->id);
            }
        }, function () {
            // 超出处理频率上限，延迟 60s 再执行
            $this->release(60);
        });
}

Redis::throttle('posts.views.increment')
    ->allow(60)->every(60)
    ->then(function () {
        // 队列任务正常处理逻辑
        if ($this->post->increment('views')) {
            Redis::zincrby('popular_posts', 1, $this->post->id);
        }
    }, function () {
        // 超出处理频率上限，延迟60s再执行
        $this->release(60);
    });