php扩展之Yar使用

<?php  
namespace app\index\logic;

class User extends RpcServer
{

    // 用户扩展信息
    public function userExt($ids)
    {
        // 1.验证签名
        // 2.逻辑处理
        // 3.结果返回
        return $ids;
    }

    // 用户基础信息
    public function userBase($ids)
    {
        return $ids;
    }


}

<?php
namespace app\index\logic;

/**
 * rpc基类(客户端)
 */
class RpcClient
{

    private static $signs = [
        'sign1', //不同来源
        'sign2'
    ];

    private $callBack;
    private $callNum = 0;


    /**
     * 取得签名
     * @param  $params 接口调用时的参数
     */
    protected function getSign($params,$type)
    {
        ksort($params);
        $signStr = '';
        foreach($params as $key => $val)
        {
            if(empty($val)) continue;
            $signStr .= $key.'='.$val.'&';
        }
        $signStr = rtrim($signStr,'&');
        return md5($signStr.self::$signs[$type]);
    }

   /**
     * 调用服务端接口
     * @param  $server      Api server
     * @param  $api         接口
     * @param  $params      参数
     * @param  $openSign    开启签名
     * @param  $callBack    回调
     */ 
    public function call($server,$api,$params,$openSign=false,$callBack=null)
    {
        if($openSign){
            $params['sign'] = $this->getSign($params);
        }
 
        if($callBack === null){
            $client = new \Yar_Client($server);
            return call_user_func_array([$client,$api], $params);
        }
        $this->callNum ++;
        $this->callBack = $callBack;
        return \Yar_Concurrent_Client::call($server,$api,$params,array($this, 'ApiClientCallBack'));
    }

     /**
     * 执行并发调用
     */
    public function loop()
    {
        return \Yar_Concurrent_Client::loop([$this,'callback1'],[$this,'error_callback']); 
    }

     /**
     * 并发调用回调
     * @param  $retval
     * @param  $callinfo
     */
    public function ApiClientCallBack($retval,$callinfo)
    {    
        if($callinfo === null){
            return $this->callBack($retval,$callinfo);
        }

        static $data = array();
        $data[] = $retval; //并发
        if(count($data) == $this->callNum){
            $fn = $this->callBack;
            return $this->$fn($data,$callinfo);
        }
    }

    // 
    public function callback1($retval, $callinfo)
    {
             if ($callinfo == NULL) {
                echo "现在, 所有的请求都发出去了, 还没有任何请求返回\n";
             } else {
                echo "这是一个远程调用的返回, 调用的服务名是", $callinfo["method"], 
                              ". 调用的sequence是 " , $callinfo["sequence"] , "\n";
                var_dump($retval);
             }
    }

    // 异常回调
    public function error_callback($type, $error, $callinfo)
    {
        error_log(json_encode(func_get_args() ),3,'rpc.log' );
    }

}

<?php 
namespace app\index\logic;

// 相关测试
class Test extends RpcClient
{
    public function testRpc()
    {

        $api = 'http://thinkphp.com/index/rpc/users';

        // $this->call($api,'userExt',[1,2],false,'callback');
        $this->call($api,'userBase',[3,4],false,'callback');
        $this->call($api,'userBase',[5],false,'callback');
        $this->loop();

        return false;
        // $client = new yar_client("http://thinkphp.com/index/rpc/user");
        // $ret = $client->userInfo([1,2]);
        // var_dump($ret);
    }

    // 回调数据
    public function callback($data,$callinfo)
    {
        var_dump(func_get_args());die;
        // static $a = [];
        // $a[] = json_encode(func_get_args());
        // print_r($a);
        // error_log(json_encode(func_get_args() ),3,'rpc.log' );
    }
}