[原创]Swoft源码剖析-RPC功能实现

Swoft提供了一个自建RPC(远程方法调用)实现，让你可以方便的调用其他Swoft上的服务。

RPC服务端的初始化

RPC有两种启动方式Http伴随启动和RPC单独启动。值得一提的是目前swoole的tcp服务即RPC服务，暂没有其他的tcp服务功能，所以基本上tcp相关的配置指代的就是RPC。

Http伴随启动

swoft的RPC 服务在Http服务启动时候伴随启动

//Swoft\Http\Server\Http\HttpServer.php

/**
 * Http Server
 */
class HttpServer extends AbstractServer
    /**
     * Start Server
     *
     * @throws \Swoft\Exception\RuntimeException
     */
    public function start()
    {
        //coco ...

        //根据.env配置文件Server区段的TCPABLE字段决定是否启动RPC服务
        if ((int)$this->serverSetting['tcpable'] === 1) {
            $this->registerRpcEvent();
        }
        //code ....
    }
}

Swoole监听

初始化流程即根据相关注解注册一个swoole监听

//Swoft\Http\Server\Http\HttpServer.php
    /**
     * Register rpc event, swoft/rpc-server required
     *
     * @throws \Swoft\Exception\RuntimeException
     */
    protected function registerRpcEvent()
    {
        //含有@SwooleListener且type为SwooleEvent::TYPE_PORT的Bean,即RpcEventListener
        $swooleListeners = SwooleListenerCollector::getCollector();
        if (!isset($swooleListeners[SwooleEvent::TYPE_PORT][0]) || empty($swooleListeners[SwooleEvent::TYPE_PORT][0])) {
            throw new RuntimeException("Please use swoft/rpc-server, run 'composer require swoft/rpc-server'");
        }

        //添加swoole RPC相关的tcp监听端口，使用的是.env文件中的TCP区段配置
        $this->listen = $this->server->listen($this->tcpSetting['host'], $this->tcpSetting['port'], $this->tcpSetting['type']);
        $tcpSetting = $this->getListenTcpSetting();
        $this->listen->set($tcpSetting);

        //根据RpcEventListener的相关注解添加监听处理句柄
        $swooleRpcPortEvents = $swooleListeners[SwooleEvent::TYPE_PORT][0];
        $this->registerSwooleEvents($this->listen, $swooleRpcPortEvents);
    }

由于是初版，根据@SwooleListener获取RPC监听Bean的相关处理暂时还有点生硬。
目前swoft中type为SwooleEvent::TYPE_PORT的@SwooleListener只有RpcEventListener一个，如果添加了同类Bean容易出问题，稳定版出的时候应该会有相关优化。

RPC单独启动

入口从Swoft\Http\Server\Command\ServerCommand换成Swoft\Rpc\Server\Command\RpcCommand，流程和Http大同小异，区别仅仅在于使用前者使用Swoole\Http\Server建立Http服务器后额外监听一个Tcp端口支持Rpc，后者直接使用Swoole\Server监听Tcp来支持Rpc，此处不再赘述。

RPC请求处理

RPC服务器和HTTP服务器的区别仅仅在于与客户端交互报文格式和报文所在的网络层(Swoft的RPC基于TCP层次)，运行原理基本相通，都是路由，中间件，RPC　Service(对应Http的Controller)，你完全可以以Http服务的思路去理解他。

swoole的RPC-TCP监听设置好后，RPC服务端就可以开始接受请求了。RpcEventListener的负责的工作仅仅是把收到的数据转发给\Swoft\Rpc\Server\ServiceDispatcher分发。Dispatcher会将请求传递给各个Middleware中间件，最终最终传递给HandlerAdapterMiddleware处理。

PackerMiddleware

PackerMiddleware是RPC中比较重要的一个中间件，负责将TCP请求中数据流解包和数据流封包。

<?php
//Swoft\Rpc\Server\Middleware.PackerMiddleware
namespace Swoft\Rpc\Server\Middleware;
/**
 * service packer
 *
 * @Bean()
 * @uses      PackerMiddleware
 * @version   2017年11月26日
 * @author    stelin <phpcrazy@126.com>
 * @copyright Copyright 2010-2016 swoft software
 * @license   PHP Version 7.x {@link http://www.php.net/license/3_0.txt}
 */
class PackerMiddleware implements MiddlewareInterface
{
    /**
     * packer middleware
     *
     * @param \Psr\Http\Message\ServerRequestInterface     $request
     * @param \Psr\Http\Server\RequestHandlerInterface $handler
     *
     * @return \Psr\Http\Message\ResponseInterface
     */
    public function process(ServerRequestInterface $request, RequestHandlerInterface $handler): ResponseInterface
    {
        //获取servicePacker Bean(\Swoft\Rpc\Packer\ServicePacker)用于字符串解包封包
        $packer = service_packer();
        $data   = $request->getAttribute(self::ATTRIBUTE_DATA);
        $data   = $packer->unpack($data);

        // 触发一个RpcServerEvent::BEFORE_RECEIVE事件，默认只有一个用于添加请求上下文信息的BeforeReceiveListener
        // 利用中间件触发流程关键事件的做法耦合有点高，猜测以后会调整
        App::trigger(RpcServerEvent::BEFORE_RECEIVE, null, $data);
        //替换解包后的解包到Request中，提供给后续中间件和Handler使用
        $request = $request->withAttribute(self::ATTRIBUTE_DATA, $data);

        /* @var \Swoft\Rpc\Server\Rpc\Response $response */
        $response      = $handler->handle($request);

       //为Response封包返回给RPC客户端
        $serviceResult = $response->getAttribute(HandlerAdapter::ATTRIBUTE);
        $serviceResult = $packer->pack($serviceResult);
        return $response->withAttribute(HandlerAdapter::ATTRIBUTE, $serviceResult);
    }
}

RouterMiddleware

RouterMiddleware负责根据RPC请求的method,version,interface 获取处理的RPC服务类，充当了路由的作用

<?php
//Swoft\Rpc\Server\Middleware\RouterMiddleware.php

/**
 * service router
 *
 * @Bean()
 * @uses      RouterMiddleware
 * @version   2017年11月26日
 * @author    stelin <phpcrazy@126.com>
 * @copyright Copyright 2010-2016 swoft software
 * @license   PHP Version 7.x {@link http://www.php.net/license/3_0.txt}
 */
class RouterMiddleware implements MiddlewareInterface
{
    /**
     * get handler from router
     *
     * @param \Psr\Http\Message\ServerRequestInterface     $request
     * @param \Psr\Http\Server\RequestHandlerInterface $handler
     *
     * @return \Psr\Http\Message\ResponseInterface
     */
    public function process(ServerRequestInterface $request, RequestHandlerInterface $handler): ResponseInterface
    {
        // service data
        $data = $request->getAttribute(PackerMiddleware::ATTRIBUTE_DATA);

        $method    = $data['method']??"";
        $version   = $data['version']??"";
        $interface = $data['interface']??"";

        /* @var \Swoft\Rpc\Server\Router\HandlerMapping $serviceRouter */
        $serviceRouter  = App::getBean('serviceRouter');
        //路由匹配，即向Swoft\Rpc\Server\Router\HandlerMapping->$routes获取RPC服务信息
        $serviceHandler = $serviceRouter->getHandler($interface, $version, $method);

        // deliver service data
        $request = $request->withAttribute(self::ATTRIBUTE, $serviceHandler);

        return $handler->handle($request);
    }
}

swoft启动阶段会扫描并初始化注解信息(参考注解章节)，注解初始化完毕后会触发一个AppEvent::APPLICATION_LOADER事件，此时会将来自@Service的所有RPC的路由信息注册到Swoft\Rpc\Server\Router\HandlerMapping->$routes中，用于serviceRouter Bean的路由匹配。

HandlerAdapterMiddleware

HandlerAdapterMiddleware最终转发请求给HandlerAdapter处理,HandlerAdapter会使用刚刚RouterMiddleware匹配到的服务类信息转发请求并封装Response最终返回给ServiceDispatcher，ServiceDispatcher会返回TCP流给客户端然后结束本次请求。

<?php
//Swoft\Rpc\Server\Router\HandlerAdapter.php
/**
 * Service handler adapter
 * @Bean("serviceHandlerAdapter")
 */
class HandlerAdapter implements HandlerAdapterInterface
{

    /**
     * Execute service handler
     *
     * @param \Psr\Http\Message\ServerRequestInterface $request
     * @param array                                    $handler
     * @return Response
     */
    public function doHandler(ServerRequestInterface $request, array $handler): Response
    {
        // RPC方法的各个参数
        $data = $request->getAttribute(PackerMiddleware::ATTRIBUTE_DATA);
        $params = $data['params'] ?? [];
        
        //路由解析出来的，处理该请求的服务Bean和方法
        list($serviceClass, $method) = $handler;
        $service = App::getBean($serviceClass);

        // execute handler with params
        $response = PhpHelper::call([$service, $method], $params);
        $response = ResponseHelper::formatData($response);

        // 构造Response返回客户端
        if (! $response instanceof Response) {
            $response = (new Response())->withAttribute(self::ATTRIBUTE, $response);
        }

        return $response;
    }
}

RPC客户端的实现

在Bean的属性中声明@Reference，swoft即会根据@var声明的类型注入相应的RPC客户端实例。

    /**
     * @Reference(name="user")
     *
     * @var DemoInterface
     */
    private $demoService;

依赖注入的实现会专门另外用一篇文章单独解释，这里先看看RPC客户端的相关代码。

远程代理

namespace Swoft\Rpc\Client\Service;

/**
 * The proxy of service
 */
class ServiceProxy
{
    /**
     * @param string $className
     * @param string $interfaceClass
     */
    public static function   (string $className, string $interfaceClass)
    {
        $reflectionClass   = new \ReflectionClass($interfaceClass);
        $reflectionMethods = $reflectionClass->getMethods(\ReflectionMethod::IS_PUBLIC);

        $template = "class $className extends \\Swoft\\Rpc\\Client\\Service implements {$interfaceClass} {";

        //\Swoft\Rpc\Client\Service::class
        // the template of methods
        $template .= self::getMethodsTemplate($reflectionMethods);
        $template .= "}";
            
        eval($template);
    }
    //code ...
}

和AOP一样，原理一样是使用了动态代理，更具体的说法是动态远程代理。
RPC动态客户端类实现了客户端声明的Interface类型(如DemoInterface)并继承了\Swoft\Rpc\Client\Service类。
动态类的实现很简单，对于接口显式声明的方法，实际上都是调用\Swoft\Rpc\Client\Service->call()方法。

interface DemoInterface
{
    /**
     * @param array $ids
     * @return array
     */
    public function getUsers(array $ids);
}

class 动态生成RPC客户端类 extends \Swoft\Rpc\Client\Service implements \App\Lib\DemoInterface { 
    public function getUsers ( array  $ids  ) {
        $params = func_get_args();
        return $this->call('getUsers', $params);
    }
    //code ...
}

对于自动生成的defer方法，则是通过魔术方法__call(),调用\Swoft\Rpc\Client\Service->deferCall()

    /**
     * @param string $name
     * @param array  $arguments
     *
     * @return ResultInterface
     * @throws RpcClientException
     */
    function __call(string $name, array $arguments)
    {
        $method = $name;
        $prefix = self::DEFER_PREFIX;//'defer'
        if (strpos($name, $prefix) !== 0) {
            throw new RpcClientException(sprintf('the method of %s is not exist! ', $name));
        }

        if ($name == $prefix) {
            $method = array_shift($arguments);
        } elseif (strpos($name, $prefix) === 0) {
            $method = lcfirst(ltrim($name, $prefix));
        }

        return $this->deferCall($method, $arguments);
    }

我们这里只看具有代表性的call()方法,deferCall()大致相同。
RPC客户端动态类的本质是将客户端的参数和接口信息根据swoft自己的格式传递给RPC服务端，然后将服务器返回的数据解包取出返回值返回给RPC的调用者，对外伪装成一个普通的对象，屏蔽远程调用操作。

// Swoft\Rpc\Client\Service.php
    /**
     * Do call service
     *
     * @param string $func
     * @param array  $params
     *
     * @throws \Throwable
     * @return mixed
     */
    public function call(string $func, array $params)
    {
        $profileKey = $this->interface . '->' . $func;
        //根据@reference的fallback属性获取降级处理句柄，在RPC服务调用失败的时候可以会使用fallback句柄代替
        $fallback   = $this->getFallbackHandler($func);
        try {
            $connectPool    = $this->getPool();
            $circuitBreaker = $this->getBreaker();

            /* @var $client AbstractServiceConnection */
            $client = $connectPool->getConnection();
            //数据封包，和RPC服务端一致
            $packer   = service_packer();
            $type     = $this->getPackerName();
            $data     = $packer->formatData($this->interface, $this->version, $func, $params);
            $packData = $packer->pack($data, $type);

            //通过熔断器调用接口
            $result = $circuitBreaker->call([$client, 'send'], [$packData], $fallback);
            if ($result === null || $result === false) {
                return null;
            }

            //和defercall不一致这里直接收包，解包
            App::profileStart($profileKey);
            $result = $client->recv();
            App::profileEnd($profileKey);
            $connectPool->release($client);

            App::debug(sprintf('%s call %s success, data=%', $this->interface, $func, json_encode($data, JSON_UNESCAPED_UNICODE)));
            $result = $packer->unpack($result);
            $data   = $packer->checkData($result);
        } catch (\Throwable $throwable) {
            if (empty($fallback)) {
                throw $throwable;
            }
            //RPC调用失败则调用降级句柄，代替实际RPC服务直接返回
            $data = PhpHelper::call($fallback, $params);
        }

        return $data;
    }

熔断器

熔断器的swoft-RPC的另一重要概念，RPC的所有请求都通过熔断器发送。
熔断器使用状态模式实现,熔断器有开启,半开,关闭 3种状态，不同状态下熔断器会持有不同的状态实例，状态根据RPC调用情况切换，熔断器根据持有状态实例的不同，行为也有所不同。

熔断器关闭状态策略

<?php
//Swoft\Sg\Circuit\CloseState.php 
/**
 * close状态的熔断器，对所有RPC调用都通过协程客户端发送到RPC服务器
 *  关闭状态及切换
 * 1. 重置failCounter=0 successCount=0
 * 2. 操作失败，failCounter计数
 * 3. 操作失败一定计数，切换为open开启状态
 */
class CloseState extends CircuitBreakerState
{
    /**
     * 熔断器调用
     *
     * @param mixed $callback 回调函数
     * @param array $params 参数
     * @param mixed $fallback 失败回调
     *
     * @return mixed 返回结果
     */
    public function doCall($callback, $params = [], $fallback = null)
    {
        list($class, $method) = $callback;

        try {
            if ($class == null) {
                throw new \Exception($this->circuitBreaker->serviceName . "服务,连接建立失败(null)");
            }

            if ($class instanceof Client && $class->isConnected() == false) {
                throw new \Exception($this->circuitBreaker->serviceName . "服务,当前连接已断开");
            }
            //调用swoole协程客户端的send()方法发送数据
            $data = $class->$method(...$params);
        } catch (\Exception $e) {
            //递增失败计数
            if ($this->circuitBreaker->isClose()) {
                $this->circuitBreaker->incFailCount();
            }

            App::error($this->circuitBreaker->serviceName . "服务，当前[关闭状态]，服务端调用失败，开始服务降级容错处理，error=" . $e->getMessage());
            //RPC调用失败则使用降级接口 
            $data = $this->circuitBreaker->fallback($fallback);
        }
        
        //失败次数过线则切换状态
        $failCount = $this->circuitBreaker->getFailCounter();
        $switchToFailCount = $this->circuitBreaker->getSwitchToFailCount();
        if ($failCount >= $switchToFailCount && $this->circuitBreaker->isClose()) {
            App::trace($this->circuitBreaker->serviceName . "服务，当前[关闭状态]，服务失败次数达到上限，开始切换为开启状态，failCount=" . $failCount);
            $this->circuitBreaker->switchToOpenState();
        }

        App::trace($this->circuitBreaker->serviceName . "服务，当前[关闭状态]，failCount=" . $this->circuitBreaker->getFailCounter());
        return $data;
    }
}

熔断器开启状态策略

<?php
\\Swoft\Sg\Circuit\OpenState .php;
/**
 * open状态的熔断器，对所有RPC调用都使用降级句柄代替
 * 开启状态及切换(open)
 * 1. 重置failCounter=0 successCounter=0
 * 2. 请求立即返回错误响应
 * 3. 定时器一定时间后切换为半开状态(open)
 */
class OpenState extends CircuitBreakerState
{
    /**
     * 熔断器调用
     *
     * @param mixed $callback 回调函数
     * @param array $params 参数
     * @param mixed $fallback 失败回调
     *
     * @return mixed 返回结果
     */
    public function doCall($callback, $params = [], $fallback = null)
    {
        $data = $this->circuitBreaker->fallback();

        App::trace($this->getServiceName() . "服务，当前[开启状态]，执行服务fallback服务降级容错处理");
        $nowTime = time();

        if ($this->circuitBreaker->isOpen()
            && $nowTime > $this->circuitBreaker->getSwitchOpenToHalfOpenTime()
        ) {
            $delayTime = $this->circuitBreaker->getDelaySwitchTimer();

            // swoole定时器不是严格的，3s容错时间 ，定时切换状态的半开
            $switchToHalfStateTime = $nowTime + ($delayTime / 1000) + 3;
            App::getTimer()->addAfterTimer('openState', $delayTime, [$this, 'delayCallback']);
            $this->circuitBreaker->setSwitchOpenToHalfOpenTime($switchToHalfStateTime);

            App::trace($this->getServiceName() . "服务，当前[开启状态]，创建延迟触发器，一段时间后状态切换为半开状态");
        }

        return $data;
    }

}

熔断器半开状态策略

半开熔断器是熔断器关闭状态和熔断器开启状态的过度状态，半开熔断器的所有RPC调用都是加锁的，连续成功或者连续失败到阈值后会切换到关闭状态或者开启状态，代码类似，此处不再累述，有兴趣的读者可以自行研究。

Swoft源码剖析系列目录：https://www.jianshu.com/p/2f679e0b4d58