   'phpDocumentor\Descriptor\FileDescriptor * hash b56a98dd4dbb518df89be62d858c3030 * pathProxy/RedisProxyCluster.php	 * source/z<?php
/**
 * 分布式结构Redis代理
 *
 * @author camera360_server@camera360.com
 * @copyright Chengdu pinguo Technology Co.,Ltd.
 */

namespace PG\MSF\Proxy;

use Exception;
use Flexihash\Flexihash;
use Flexihash\Hasher\Md5Hasher;
use PG\MSF\Pools\RedisAsynPool;
use PG\MSF\Helpers\Context;

/**
 * Class RedisProxyCluster
 * @package PG\MSF\Proxy
 */
class RedisProxyCluster extends Flexihash implements IProxy
{
    /**
     * @var string 代理标识，它代表一个Redis集群
     */
    private $name;

    /**
     * @var array 连接池列表 key=连接池名称, value=权重
     */
    private $pools;

    /**
     * @var array 通过探活检测的连接池列表
     */
    private $goodPools = [];

    /**
     * @var mixed|string key前缀
     */
    private $keyPrefix = '';

    /**
     * @var bool|mixed 是否将key散列后储存
     */
    private $hashKey = false;
    /**
     * @var bool 随机选择一个redis，一般用于redis前面有twemproxy等代理，每个代理都可以处理请求，随机即可
     */
    private $isRandom = false;

    /**
     * RedisProxyCluster constructor.
     *
     * @param string $name 代理标识
     * @param array $config 代理配置数组
     */
    public function __construct(string $name, array $config)
    {
        $this->name      = $name;
        $this->pools     = $config['pools'];
        $this->keyPrefix = $config['keyPrefix'] ?? '';
        $this->hashKey   = $config['hashKey'] ?? false;
        $this->isRandom  = $config['random'] ?? false;
        $hasher          = $config['hasher'] ?? Md5Hasher::class;
        $hasher          = new $hasher;

        try {
            parent::__construct($hasher);
            $this->startCheck();
            if (empty($this->goodPools)) {
                throw new Exception('No redis server can write in cluster');
            } else {
                foreach ($this->goodPools as $pool => $weight) {
                    $this->addTarget($pool, $weight);
                }
            }
        } catch (Exception $e) {
            writeln('Redis Proxy ' . $e->getMessage());
        }
    }

    /**
     * 检测可用的连接池
     *
     * @return $this
     */
    public function startCheck()
    {
        $this->syncCheck();
        return $this;
    }

    /**
     * 启动时同步检测可用的连接池
     *
     * @return $this
     */
    private function syncCheck()
    {
        $this->goodPools = [];

        foreach ($this->pools as $pool => $weight) {
            try {
                $poolInstance = getInstance()->getAsynPool($pool);
                if (!$poolInstance) {
                    $poolInstance = new RedisAsynPool(getInstance()->config, $pool);
                    getInstance()->addAsynPool($pool, $poolInstance, true);
                }

                if ($poolInstance->getSync()->set('msf_active_cluster_check_' . gethostname(), 1, 5)) {
                    $this->goodPools[$pool] = $weight;
                } else {
                    $host = getInstance()->getAsynPool($pool)->getSync()->getHost();
                    $port = getInstance()->getAsynPool($pool)->getSync()->getPort();
                    getInstance()->getAsynPool($pool)->getSync()->connect($host, $port, 0.05);
                }
            } catch (\Exception $e) {
                writeln('Redis Proxy' . $e->getMessage() . "\t {$pool}");
            }
        }
    }

    /**
     * 发送异步Redis请求
     *
     * @param string $method Redis指令
     * @param array $arguments Redis指令参数
     * @return array|bool|mixed
     */
    public function handle(string $method, array $arguments)
    {
        /**
         * @var Context $arguments[0]
         */
        try {
            if ($this->isRandom) {
                return $this->random($method, $arguments);
            }

            if ($method === 'evalMock') {
                return $this->evalMock($arguments);
            } else {
                $key = $arguments[1];
                //单key操作
                if (!is_array($key)) {
                    return $this->single($method, $key, $arguments);
                    // 批量操作
                } else {
                    return $this->multi($method, $key, $arguments);
                }
            }
        } catch (\Exception $e) {
            return false;
        }
    }

    /**
     * 执行Redis evalMock指令
     *
     * @param array $arguments evalMock指令参数
     * @return array
     */
    public function evalMock(array $arguments)
    {
        $args         = $arguments[2];
        $numKeys      = $arguments[3];
        $keys         = array_slice($args, 0, $numKeys);
        $evalMockArgs = array_slice($args, $numKeys);
        $arrRedis     = $index2Key = [];

        if (empty($keys)) {
            //如果没有设置缓存key，则连接所有的实例
            $arrRedis = $this->getAllTargets();
        } else {
            //根据脚本中用到的key计算出需要连接哪些实例
            foreach ($keys as $key) {
                $key = $this->generateUniqueKey($key);
                $redisPoolName = $this->lookup($key);

                $index = array_search($redisPoolName, $arrRedis, true);
                if ($index === false) {
                    $index = count($arrRedis);
                    $arrRedis[] = $redisPoolName;
                }
                $index2Key[$index][] = $key;
            }
        }

        $ret = [];
        foreach ($arrRedis as $index => $redisPoolName) {
            if (!isset(RedisProxyFactory::$redisCoroutines[$redisPoolName])) {
                if (getInstance()->getAsynPool($redisPoolName) == null) {
                    return [];
                }
                RedisProxyFactory::$redisCoroutines[$redisPoolName] = getInstance()->getAsynPool($redisPoolName)->getCoroutine();
            }
            $redisPoolCoroutine = RedisProxyFactory::$redisCoroutines[$redisPoolName];

            $arrKeys = empty($index2Key) ? $index2Key : $index2Key[$index];
            $res     = $redisPoolCoroutine->evalMock($arguments[0], $arguments[1], array_merge($arrKeys, $evalMockArgs), count($arrKeys));

            if ($res instanceof \Generator) {
                $ret[] = $res;
            }
        }

        foreach ($ret as $k => $item) {
            $ret[$k] = $item;
        }

        return $ret;
    }

    /**
     * 生成唯一Redis Key
     *
     * @param string $key Key
     * @return string
     */
    private function generateUniqueKey(string $key)
    {
        return $this->hashKey ? md5($this->keyPrefix . $key) : $this->keyPrefix . $key;
    }

    /**
     * 随机策略
     *
     * @param string $method Redis指令
     * @param array $arguments Redis指令参数
     * @return bool
     */
    private function random(string $method, array $arguments)
    {
        $redisPoolName = array_rand($this->goodPools);

        if (!isset(RedisProxyFactory::$redisCoroutines[$redisPoolName])) {
            if (getInstance()->getAsynPool($redisPoolName) == null) {
                return false;
            }
            RedisProxyFactory::$redisCoroutines[$redisPoolName] = getInstance()->getAsynPool($redisPoolName)->getCoroutine();
        }
        $redisPoolCoroutine = RedisProxyFactory::$redisCoroutines[$redisPoolName];

        if ($method === 'cache') {
            $result = $redisPoolCoroutine->$method(...$arguments);
        } else {
            $result = $redisPoolCoroutine->__call($method, $arguments);
        }

        return $result;
    }

    /**
     * 单key指令
     *
     * @param string $method Redis指令
     * @param string $key Redis Key
     * @param array $arguments Redis指令参数
     * @return mixed
     */
    private function single(string $method, string $key, array $arguments)
    {
        $redisPoolName = $this->lookup($this->generateUniqueKey($key));

        if (!isset(RedisProxyFactory::$redisCoroutines[$redisPoolName])) {
            if (getInstance()->getAsynPool($redisPoolName) == null) {
                return false;
            }
            RedisProxyFactory::$redisCoroutines[$redisPoolName] = getInstance()->getAsynPool($redisPoolName)->getCoroutine();
        }
        $redisPoolCoroutine = RedisProxyFactory::$redisCoroutines[$redisPoolName];

        if ($method === 'cache') {
            $result = $redisPoolCoroutine->$method(...$arguments);
        } else {
            $result = $redisPoolCoroutine->__call($method, $arguments);
        }

        return $result;
    }

    /**
     * 批量多key指令
     *
     * @param string $method Redis指令
     * @param array $key Redis Key列表
     * @param array $arguments Redis指令参数
     * @return array|bool
     */
    private function multi(string $method, array $key, array $arguments)
    {
        $opArr = [];

        if (in_array(strtolower($method), ['mset', 'msetnx'])) {
            foreach ($key as $k => $v) {
                $redisPoolName = $this->lookup($this->generateUniqueKey($k));
                $opArr[$redisPoolName][$k] = $v;
            }

            $opData = $this->dispatch($opArr, $method, $arguments);

            foreach ($opData as $op) {
                $result = $op;
                if ($result !== 'OK') {
                    return false;
                }
            }

            return true;
        } else {
            $retData = [];
            foreach ($key as $k) {
                $redisPoolName = $this->lookup($this->generateUniqueKey($k));
                $opArr[$redisPoolName][] = $k;
            }

            $opData = $this->dispatch($opArr, $method, $arguments);

            foreach ($opData as $redisPoolName => $op) {
                $keys = $opArr[$redisPoolName];
                $values = $op;
                if (is_array($values)) { //$values有可能超时返回false
                    $retData = array_merge($retData, array_combine($keys, $values));
                }
            }

            return $retData;
        }
    }

    /**
     * 请求分发
     *
     * @param array $opArr 相应Redis连接池的所有请求
     * @param string $method Redis指令
     * @param array $arguments Redis指令参数
     * @return array
     */
    protected function dispatch(array $opArr, string $method, array $arguments)
    {
        $opData = [];
        foreach ($opArr as $redisPoolName => $op) {
            if (!isset(RedisProxyFactory::$redisCoroutines[$redisPoolName])) {
                if (getInstance()->getAsynPool($redisPoolName) == null) {
                    return [];
                }
                RedisProxyFactory::$redisCoroutines[$redisPoolName] = getInstance()->getAsynPool($redisPoolName)->getCoroutine();
            }
            $redisPoolCoroutine = RedisProxyFactory::$redisCoroutines[$redisPoolName];

            if ($method === 'cache') {
                $opData[$redisPoolName] = $redisPoolCoroutine->$method(...[$arguments[0], $op]);
            } else {
                $opData[$redisPoolName] = $redisPoolCoroutine->__call($method, [$arguments[0], $op]);
            }
        }

        return $opData;
    }

    /**
     * 用户定时检测
     *
     * @return bool
     */
    public function check()
    {
        $this->goodPools = getInstance()->sysCache->get($this->name) ?? [];
        if (!$this->goodPools) {
            return false;
        }

        $nowPools = $this->getAllTargets();
        $newPools = array_keys($this->goodPools);
        $loses    = array_diff($nowPools, $newPools);

        if (!empty($loses)) {
            foreach ($loses as $lost) {
                $this->removeTarget($lost);
            }
            writeln('Redis Proxy Remove ( ' . implode(',', $loses) . ' ) from Cluster');
        }

        $adds = array_diff($newPools, $nowPools);
        if (!empty($adds)) {
            foreach ($adds as $add) {
                $this->addTarget($add, $this->pools[$add]);
            }
            writeln('Redis Proxy Add ( ' . implode(',', $adds) . ' ) into Cluster');
        }

        return true;
    }
}
 * namespaceAliases#phpDocumentor\Descriptor\Collection * items	Exception
\Exception	Flexihash\Flexihash\Flexihash	Md5Hasher\Flexihash\Hasher\Md5HasherRedisAsynPool\PG\MSF\Pools\RedisAsynPoolContext\PG\MSF\Helpers\Context * includes	  * constants	  * functions	 
 * classes	\PG\MSF\Proxy\RedisProxyCluster(phpDocumentor\Descriptor\ClassDescriptor	 * parent * implements	\PG\MSF\Proxy\IProxy * abstract * final	  * properties	name+phpDocumentor\Descriptor\PropertyDescriptor" * types 
 * default 	 * static * visibilityprivate * fqsen%\PG\MSF\Proxy\RedisProxyCluster::name * name  * namespace 
 * package
 * summary * description * fileDescriptor  * line * tags	var	 *phpDocumentor\Descriptor\Tag\VarDescriptor * variableName"	 .phpDocumentor\Descriptor\Type\StringDescriptor )1-)代理标识，它代表一个Redis集群	 * errors	 6	  * inheritedElement pools!"" # $%&'&\PG\MSF\Proxy\RedisProxyCluster::pools)8* +,-. /0	1	 23"	 3phpDocumentor\Descriptor\Type\UnknownTypeDescriptor)array)1-1连接池列表 key=连接池名称, value=权重6	 6	 7 	goodPools!"" #array()$%&'*\PG\MSF\Proxy\RedisProxyCluster::goodPools)=* +,-. /$0	1	 23"	 :);)1-$通过探活检测的连接池列表6	 6	 7 	keyPrefix!"" #''$%&'*\PG\MSF\Proxy\RedisProxyCluster::keyPrefix)A* +,-. /)0	1	 23"	 :)mixed4 )1-	key前缀6	 6	 7 hashKey!"" #false$%&'(\PG\MSF\Proxy\RedisProxyCluster::hashKey)F* +,-. /.0	1	 23"	 /phpDocumentor\Descriptor\Type\BooleanDescriptor :)D)1-是否将key散列后储存6	 6	 7 isRandom!"" #G$%&')\PG\MSF\Proxy\RedisProxyCluster::isRandom)K* +,-. /20	1	 23"	 I )1-y随机选择一个redis，一般用于redis前面有twemproxy等代理，每个代理都可以处理请求，随机即可6	 6	 7 
 * methods	__construct)phpDocumentor\Descriptor\MethodDescriptor"$%public * arguments	$name+phpDocumentor\Descriptor\ArgumentDescriptor	 * method"d"	 4 #  * byReference * isVariadic')S* +,-代理标识. / 0	 6	 7 $configTU"d"	 :);# VW')Y* +,-代理配置数组. / 0	 6	 7 '.\PG\MSF\Proxy\RedisProxyCluster::__construct())O* +,RedisProxyCluster constructor.-. /:0	param	 ,phpDocumentor\Descriptor\Tag\ParamDescriptor3S""h)]-X6	 ^3Y""p)]-Z6	 return	 6	 7 
startCheckP"$%QR	 '-\PG\MSF\Proxy\RedisProxyCluster::startCheck())`* +,检测可用的连接池-. /X0	_	 -phpDocumentor\Descriptor\Tag\ReturnDescriptor"	 :)$this)_-6	 ]	 6	 7 	syncCheckP"$%&R	 ',\PG\MSF\Proxy\RedisProxyCluster::syncCheck())e* +,'启动时同步检测可用的连接池-. /c0	_	 c"	 :)d)_-6	 ]	 6	 7 handleP"$%QR	$methodTU""	 4 # VW')i* +,-Redis指令. / 0	 6	 7 
$argumentsTU""	 :);# VW')k* +,-Redis指令参数. / 0	 6	 7 ')\PG\MSF\Proxy\RedisProxyCluster::handle())h* +,发送异步Redis请求-. /0	]	 ^3i"")]-j6	 ^3k"")]-l6	 _	 c"	 :);I :)D)_-6	 6	 7 evalMockP"$%QR	kTU""	 :);# VW')k* +,-evalMock指令参数. / 0	 6	 7 '+\PG\MSF\Proxy\RedisProxyCluster::evalMock())o* +,执行Redis evalMock指令-. /0	]	 ^3k"")]-p6	 _	 c"	 :);)_-6	 6	 7 generateUniqueKeyP"$%&R	$keyTU""	 4 # VW')t* +,-Key. / 0	 6	 7 '4\PG\MSF\Proxy\RedisProxyCluster::generateUniqueKey())s* +,生成唯一Redis Key-. /0	]	 ^3t"")]-u6	 _	 c"	 4 )_-6	 6	 7 randomP"$%&R	iTU#"	 4 # VW')i* +,-j. / 0	 6	 7 kTU#"	 :);# VW')k* +,-l. / 0	 6	 7 ')\PG\MSF\Proxy\RedisProxyCluster::random())x* +,随机策略-. /0	]	 ^3i"#)]-j6	 ^3k"#)]-l6	 _	 c"	 I )_-6	 6	 7 singleP"$%&R	iTU#/"	 4 # VW')i* +,-j. / 0	 6	 7 tTU#/"	 4 # VW')t* +,-	Redis Key. / 0	 6	 7 kTU#/"	 :);# VW')k* +,-l. / 0	 6	 7 ')\PG\MSF\Proxy\RedisProxyCluster::single()){* +,单key指令-. /0	]	 ^3i"#3)]-j6	 ^3t"#;)]-|6	 ^3k"#C)]-l6	 _	 c"	 :)D)_-6	 6	 7 multiP"$%&R	iTU#a"	 4 # VW')i* +,-j. / 0	 6	 7 tTU#a"	 :);# VW')t* +,-Redis Key列表. / 0	 6	 7 kTU#a"	 :);# VW')k* +,-l. / 0	 6	 7 '(\PG\MSF\Proxy\RedisProxyCluster::multi())* +,批量多key指令-. /#0	]	 ^3i"#e)]-j6	 ^3t"#m)]-6	 ^3k"#u)]-l6	 _	 c"	 :);I )_-6	 6	 7 dispatchP"$%	protectedR	$opArrTU#"	 :);# VW')* +,-#相应Redis连接池的所有请求. / 0	 6	 7 iTU#"	 4 # VW')i* +,-j. / 0	 6	 7 kTU#"	 :);# VW')k* +,-l. / 0	 6	 7 '+\PG\MSF\Proxy\RedisProxyCluster::dispatch())* +,请求分发-. /T0	]	 ^3"#)]-6	 ^3i"#)]-j6	 ^3k"#)]-l6	 _	 c"	 :);)_-6	 6	 7 checkP"$%QR	 '(\PG\MSF\Proxy\RedisProxyCluster::check())* +,用户定时检测-. /o0	_	 c"	 I )_-6	 ]	 6	 7  * usedTraits	 ')RedisProxyCluster*\PG\MSF\Proxy+PG\MSF\Proxy,Class RedisProxyCluster-." /0	package	 &phpDocumentor\Descriptor\TagDescriptor)-6	 
subpackage	 6	 7  * interfaces	 	 * traits	 
 * markers	 ')RedisProxyCluster.php* +Default,分布式结构Redis代理-. / 0	author	 -phpDocumentor\Descriptor\Tag\AuthorDescriptor)-camera360_server@camera360.com6	 	copyright	 )-"Chengdu pinguo Technology Co.,Ltd.6	 	 )-6	 	 6	 7 