Maksim

Curl 请求引发一场血案：利用 Redis 实现熔断器

PHP , 实战2482 字

在前段时间，我负责的推送中台出现了一个非常严重的 Bug，差点造成线上服务宕机。

我简单来描述一下当时的场景。

  1. 我们使用个推来来推送我们的 APP 通知，以及用户画像，用户分析等等。
  2. 每一次在 app 启动都会访问一个接口来判断用户信息，然后绑定标签，这个动作会触发 curl 函数访问个推的 API。
  3. 结果悲剧了，个推北京机房宕机。

我们当时的主要技术栈是 PHP，我们线上 curl 请求默认超时时间是 60 秒，这就导致我们的数据库链接近乎被打满。

由于 PHP-FPM 在脚本执行完成后才会释放进程，这也包括其中的资源，curl 请求的默认时间是 60 秒，这样也就意味着在 60 秒内，该进程也会被一直阻塞，直到超时为止，而 PDO 的生命周期是跟着 FPM 走的，除非调用手动 close() 这样才会关闭。

MySQL 连接断开也是有默认超时时间的，如果在超时时间范围内，客户端没有主动关闭，该连接会一直存在，但是 MySQL 的连接数是有限的，如果连接被打满那么新的连接将无法建，整个服务 GG 。

如果这个时候再有流量进来，那么就只有宕机一条死路了，于是立即找到运维修改默认配置，然后重启 php-fpm 释放 MYSQL 连接。

处理方式以及断路器

经过这一次的线上事故，总结出了一点，对于任何第三方提供的接口，都不要完全信赖！，都要去考虑如果对方出现问题，我这里应该如何操作！

经过复盘我们当时存在以下问题：

  1. 环境配置不透明，开发基本不关注线上配置。
  2. 没有考虑第三方失败该如何进行处理。
  3. 架构设计不完善，此类功能完全也可以异步化。

第一点是由于工作经验不足或者缺乏沟通导致的，很好解决，当这个事情解决之后，团队对线上的配置情况和运维一起进行了一次梳理。

第二点，我们选择了熔断器，首先说一下什么是熔断器。

第三点，对业务进行了梳理，将强依赖接口改成了消息队列进行异步处理。

de8a07841802fdcd985039e8afb1e58b

图 1.1 熔断器

如果你看过自己家中的电闸开关的话就会看到上图这个东西， 当电流正常情况下熔断器是关闭状态电流就可以正常通过，如果电流负载过大为了保护家用电器就会开启，然后关闭电源保证电器安全。

在我们的代码中熔断器其实就是一个设计模式，我们可以通过很简单的代码就可以实现熔断器。

图 1.2 无熔断器流程图

图 1.2 中是在没有熔断器我们在开发这类业务的时候，可能会对第三方服务进行 try catch 来判断异常，如果有异常在反复几次调用，多次尝试失败后，可能就直接返回报错，提供有损的服务，像上面我提到的那场事故中这个问题就会被放大，因为 60 秒后才会触发超时，也就意味着，如果重试三次那么该接口的耗时就会增加到 3 分钟，正常情况下是不会有业务接口能够接收三分钟这样耗时的。

当我们加上熔断器后，那么流程就会变成图1.3 那样。

图 1.3

引入断路器后，不再是 app server 直接访问第三方服务，而是将访问委托给 fuse 熔断器来进行，断路器判断当前的开闭状态来判断是否要发起请求。

  1. 如果是处于关闭状态，也就是第三方服务可以正常访问，那就直接访问第三方服务，
  2. 如果熔断器处于开启状态，那么正面目前第三方服务不可用，则直接调用备用函数，这也是微服务中一个重要的概念——服务降级。

利用 Redis Sorted Set 实现简单的熔断器

由于我们使用的 PHP-FPM，这也就导致我们的 PHP 代码是无法常驻内存的，这个时候，就需要使用 Redis 来进行记数，在数据结构方面我们选择使用有序集合。

keyscoremember
circuit10pushMessage2User
circuit1pushMessage2List

score 用来存放失败的次数，member 用来存放失败的函数。

搭建骨架

我们先来按照上一小节的内容先将熔断器的挂架代码打起来。

<?php

class CircuitBreaker
{
    private $zSetKey = 'circuit';

    public function invoke(object $class, string $method, array $params, callable $fallback)
    {
        try {
            return $class->$method(...$params);

        } catch (Throwable $exception) {
            return $fallback(); //函数降级
        }
    }
}

CircuitBreaker 就是我们的熔断器类。

其中设计了一个 invoke 方法，这个方法就是上面讲到的代理请求，我们来通过 invoke 方法来让熔断器代理我们的请求。

  • $class: 我们首先要将类实例化后传入到 invoke中，这样的好处是进行依赖翻转，让熔断器可以通用。
  • $method: 是该类需要执行的方法。
  • $params: 是执行该方法时候需要传入的参数。
  • $callback:是一个闭包，当熔断器被打开，或者出现异常的时候，我们需要对服务进行降级。

下面是如何调用。

<?php

require 'CircuitBreaker.php';

class Pusher
{
    public function pushMessage2User($username, $message): bool
    {
        echo "使用 Pusher 进行发送 $username [$message]";
        return true;
    }
}

$c = new CircuitBreaker();
$messageData = ['maksim', 'hello world!'];
$pusher = new Pusher();
$c->invoke($pusher, 'pushMessage2User', $messageData, function () use ($messageData) {
    echo "使用站内信进行发送 $messageData[0] $messageData[1]";
    return true;
});

这个时候就相当于是一个简单的代理模式，由 CircuitBreaker 去执行 Pusher，并且捕获异常，如果出现异常就用降级函数。

在这里，我们模拟发送通知的业务场景，如果 Pusher 类发送失败了，那么我们就降级使用站内信的方式进行发送，当然也可以扔到一个队列里面，等业务恢复后在开启异步消息队列将消息推送出去。

为熔断器添加计数器

接下里，我们对 CircuitBreaker 进行修改，增加错误计数。

public function invoke(object $class, string $method, array $params, callable $fallback)
    {
        try {
            return $class->$method(...$params);
            
        } catch (Throwable $exception) {
            $member = get_class($class) . '_' . $method;
            $this->redis->zIncrBy($this->zSetKey, 1, $member);

            return $fallback(); //函数降级
        }
    }

这里的代码非常简单，当代码发生异常后，我们直接增加 redis 计数器，下面我们使用 Redis 的延迟队列来实现熔断器开关状态的切换。

首先，我们需要设置失败的阀值，当失败次数达到阀值后不再调用原函数。

public $failCount = 3;

private function isFail($member) {
    if ($this->redis->zScore($this->zSetKey, $member) >= $this->failCount ) {
        return true;
    }
    return false;
}
public function invoke(object $class, string $method, array $params, callable $fallback)
    {
        $member = get_class($class) . '_' . $method;
        try {
            if ($this->isFail($member)) {
                return $fallback();
            }
            return $class->$method(...$params);
            
        } catch (Throwable $exception) {
   
            $this->redis->zIncrBy($this->zSetKey, 1, $member);

            return $fallback(); //函数降级
        }
    }

实现熔断器的三种状态

现在我们就实现了熔断器的打开，接下来，我们需要实现熔断器的关闭，但是只有这两种状态熔断器是无法正常运作的，在日常生活中，当熔断器跳闸后，我们判断没有问题后会手动去尝试开启熔断器，同样在代码中，我们也需要一个这个样的过程，我们需要让熔断器再去尝试着去访问原来的方法，如果满足可关闭状态，那么把熔断器关掉。

而在这个试探的这个状态就是半开状态。

  1. 关：调用原函数，失败后计数器+1，并返回降级函数，到达一定阀值后进入“开”状态；
  2. 开：也就是直接返回降级函数，不调用原函数，但是我们可以给失败函数一些机会，于是出现半开状态，不过需要设定一定的定时器，一定时间后进入半开状态。
  3. 半开状态：处理半开妆容是熔断器的核心，调用函数时候有一定数量的调用时进入降级函数，有一定数量进入的是真实函数。如果函数已经稳定，那么就进入关状态。

状态值的定义如下:

define("BreakerStateOpen" , 1);         // 开
define("BreakerStateClose" , 2);         // 关，这是默认值
define("BreakerStateHalfOpen" , 3);     // 半开

private funcion getState($member) {
    $getSocre = $this->redis->zScore($this->zSetKey, $member);
    if ($getScore >= $this->failCOunt {
        return BreakerStateOpen;
    }
    if ($getScore < 0) {
        return BreakerStateHalfOpen; // 如果值是 -1 则代表是半开状态
    }
   return BreakerStateClose;
}

接下来，利用定时器来完成进入半开状态，在 PHP-FPM 模式下定时器我们可以通过 Redis 实现，实现的方式有两种：

第一种：利用一个带有过期时间的 key 来完成，一旦进入“开”状态后，插入一个 key，设置过期时间，然后利用过期通知的方式来完成。

第二种：延迟队列，依然使用 Sorted Set 来完成，key 设置为 circuit_open，插入时间戳，编写一个死循环程序，反复监听。

通过这两种方式，监听过期时间，到期后把 circuit 对应的 member 的 score 设置为 -1，即代表是半开状态。

在这里我们通过延迟队列的方式来进行完成，如果不了解延迟队列可以通过《PHP 利用 Redis Sorted Set 实现延迟队列》这片文章来了解其原理。

<?php
  
while(true) {
    $members = $redis->zRangeByScore("circuit_open", "-inf", time(), ['limit' => [0, 10]]);
    
    if (count($members) > 0) {
        foreach ($members as $member) {
              $redis->zAdd('circuit', -1, $member);
        }
        
        //删掉 open key
          $redis->zRem("circuit_open", ...$members);
    }
    
    usleep(500 * 1000);//休眠 500 毫秒
}

这样一来，我们就可以使用死循环程序来不断的设置半开状态了。

接下来我们在 invoke 方法来处理，当熔断器进入到开状态时那么就设置一个定时器。

public function invoke(object $class, string $method, array $params, callable $fallback)
    {
        $member = get_class($class) . '_' . $method;
        $currentState = $this->getState($member); //获取当前状态
        try {
            if ($currentState == BreakerStateOpen)  {
                return $fallback();
            } else if ($currentState == BreakerStateHalfOpen) {
                //如果是半开状态
                if (rand(0, 100) % 2 == 0) {
                    return $fallback();
                } else {
                    $result = $class->$method(...$params);
                    // 半开状态下仍然要加计数器，目的是为了让他归 0
                    $this->redis->zIncrBy($this->zSetKey, 1, $member); 
                    return $result;
                }
            }
            return $class->$method(...$params);
            
        } catch (Throwable $exception) {
            if ($currentState == BreakerStateClose) {
                 //切换到开
                   $socre = $this->redis->zIncrBy($this->zSetKey, 1, $member);
                if ($score == $this->failCount) {
                    //增加队列
                       $this->redis->zAdd($this->zSetKey_open, time() + $this->openTime, $member);
                }
            }
            
            if ($currentState == BreakerStateHalfOpen) {
                $this->redis->zIncrBy($this->zSetKey, $this->failCount, $member); // 将熔断器重新设置为开状态
                $this->redis->zAdd($this->zSetKey_open, time() + $this->openTime, $member); //再次开启定时器
            }   
            return $fallback(); //函数降级
        }
    }

在这段代码中，我们重点关注第 10 行到第 16 行，这里“随机”让其访问真实函数，如果函数正常，则将计数器归 0，让其进入到关闭状态。

在第 22 行开始增加了对状态的判断，如果当前处于关闭状态，那么久打开，如果现在是半开状态，那证明半开尝试失败了，需要重新进入打开状态。

这里的半开状态的处理很简单，其实正常的熔断器还需要包括采样分析，来判断当前是否满足开合条件，同时上面的“随机”也无法控制到底有多少流量进入到正常函数，所以我们还需要增加真正的概率算法，将通过的流量控制在一个我们可以接受的范围内。

maksim
Maksim（一笑，吡罗），PHPer，Goper
OωO
开启隐私评论，您的评论仅作者和评论双方可见