添加异步redis,添加redisProxy

@param \swoole_server $serv server实例
@param int $workerId worker id
@throws Exception
766767
public function onWorkerStart($serv, $workerId)
    {
        // Worker类型
        if (!$this->isTaskWorker()) {
            if ($workerId !== null) {
                $this->processType = Macro::PROCESS_WORKER;
                getInstance()->sysTimers[] = swoole_timer_tick(2000, function ($timerId) {
                    $this->statistics();
                });
            }
        } else {
            $this->processType = Macro::PROCESS_TASKER;
        }

        parent::onWorkerStart($serv, $workerId);
        if ($this->processType == Macro::PROCESS_WORKER || $this->processType == Macro::PROCESS_TIMER) {
            $this->initAsynPools();
            $this->initRedisProxies();
            $this->initMysqlProxies();

            //注册
            $this->asynPoolManager = new AsynPoolManager(null, $this);
            $this->asynPoolManager->noEventAdd();
            foreach ($this->asynPools as $pool) {
                if ($pool) {
                    $pool->workerInit($workerId);
                    $this->asynPoolManager->registerAsyn($pool);
                }
            }

            if (!empty($this->redisProxyManager)) {
                //redis proxy监测
                getInstance()->sysTimers[] = $this->server->tick(5000, function () {
                    foreach ($this->redisProxyManager as $proxy) {
                        $proxy->check();
                    }
                });
            }

            if (!empty($this->mysqlProxyManager)) {
                // mysql proxy监测
                getInstance()->sysTimers[] = $this->server->tick(5000, function () {
                    foreach ($this->mysqlProxyManager as $proxy) {
                        $proxy->check();
                    }
                });
            }
        }
    }
336525
@Override
    public synchronized ServerBootstrap newServerBootstrap() throws Exception {
        if (serverChannelFactory == null) {
            Executor workerExecutor = executorServiceFactory.newExecutorService("worker.server");
            NioDatagramWorkerPool workerPool = new NioDatagramWorkerPool(workerExecutor, 1);
            serverChannelFactory = new UdpServerChannelFactory(workerPool, timer);

            // unshared
            channelFactories.add(serverChannelFactory);
        }

        return new ServerBootstrap(serverChannelFactory);
    }
1162740
public function dispathToWorkerId($workerId, $type, $data = null)
    {
        get_instance()->sendToOneWorker($workerId, SwooleMarco::DISPATCHER_NAME, [$type, $data], EventDispatcher::class . "::workerDispatchEventWith");
    }
1116538
public function onPipeMessage(Server $server, int $srcWorkerId, string $data): void
    {
        $this->log("worker #{$server->worker_id} received message from #$srcWorkerId, data: $data");
    }
836443
public function onConnect(SwServer $server, int $fd, int $reactorId)
    {
        // {{ log
        $this->get('logger')->info('连接接入', $server->getClientInfo($fd));
        // }}

        $this->callProtocolMethod('onConnect', $server, $fd, $fromId);
    }
780202
public function sendRequest(Worker $worker, Request $request)
    {
        $worker->setRequest($request);
        $this->reply(
            $worker,
            new ExecuteJobRequest(
                $request->getId(),
                $request->getActionName(),
                $request->getParams()
            )
        );
    }
606054
public function onWorkerStop(SwooleServer $server, $worker_id)
    {
        //不能在这里处理 coroutine ，所以不能使用 Yii::$app
        $contexts = Yii::$context->getContextData();
        foreach ($contexts as $context){
            /** @var Application $application */
            $application = $context[Context::COROUTINE_APP] ?? null;
            if(empty($application)){
                continue;
            }
            $targets = $application->getLog()->targets;
            foreach ($targets as $target) {
                $target->export();
            }
        }

    }
120906
protected function changeWorkerState(string $workerId, int $workerState) : bool
    {
        $this->workerStates[$workerId] = $workerState;
        if ($workerState === Worker::WORKER_STATE_IDLE) {
            $this->pushJobs();
        }
        return true;
    }
608971
private function markAsActive($workerID = 0)
    {
        if ($workerID == 0) {
            $query = $this->pdo->prepare(<<<SQL
UPDATE robotstxt__cache1
SET worker = NULL
WHERE base = :base AND worker = 0;
SQL
            );
            $query->bindValue('base', $this->base, \PDO::PARAM_STR);
            return $query->execute();
        }
        return true;
    }
622745
public function init() {
        if (!extension_loaded('redis')) {
            throw new exception('sessionhandlerproblem', 'error', '', null, 'redis extension is not loaded');
        }

        if (empty($this->host)) {
            throw new exception('sessionhandlerproblem', 'error', '', null,
                    '$CFG->session_redis_host must be specified in config.php');
        }

        // The session handler requires a version of Redis with the SETEX command (at least 2.0).
        $version = phpversion('Redis');
        if (!$version or version_compare($version, '2.0') <= 0) {
            throw new exception('sessionhandlerproblem', 'error', '', null, 'redis extension version must be at least 2.0');
        }

        $this->connection = new \Redis();

        $result = session_set_save_handler(array($this, 'handler_open'),
            array($this, 'handler_close'),
            array($this, 'handler_read'),
            array($this, 'handler_write'),
            array($this, 'handler_destroy'),
            array($this, 'handler_gc'));
        if (!$result) {
            throw new exception('redissessionhandlerproblem', 'error');
        }

        // MDL-59866: Add retries for connections (up to 5 times) to make sure it goes through.
        $counter = 1;
        $maxnumberofretries = 5;

        while ($counter <= $maxnumberofretries) {

            try {

                $delay = rand(100000, 500000);

                // One second timeout was chosen as it is long for connection, but short enough for a user to be patient.
                if (!$this->connection->connect($this->host, $this->port, 1, null, $delay)) {
                    throw new RedisException('Unable to connect to host.');
                }

                if ($this->auth !== '') {
                    if (!$this->connection->auth($this->auth)) {
                        throw new RedisException('Unable to authenticate.');
                    }
                }

                if (!$this->connection->setOption(\Redis::OPT_SERIALIZER, $this->serializer)) {
                    throw new RedisException('Unable to set Redis PHP Serializer option.');
                }

                if ($this->prefix !== '') {
                    // Use custom prefix on sessions.
                    if (!$this->connection->setOption(\Redis::OPT_PREFIX, $this->prefix)) {
                        throw new RedisException('Unable to set Redis Prefix option.');
                    }
                }
                if ($this->database !== 0) {
                    if (!$this->connection->select($this->database)) {
                        throw new RedisException('Unable to select Redis database '.$this->database.'.');
                    }
                }
                $this->connection->ping();
                return true;
            } catch (RedisException $e) {
                $logstring = "Failed to connect (try {$counter} out of {$maxnumberofretries}) to redis ";
                $logstring .= "at {$this->host}:{$this->port}, error returned was: {$e->getMessage()}";

                // @codingStandardsIgnoreStart
                error_log($logstring);
                // @codingStandardsIgnoreEnd
            }

            $counter++;

            // Introduce a random sleep between 100ms and 500ms.
            usleep(rand(100000, 500000));
        }

        // We have exhausted our retries, time to give up.
        return false;
    }
142653
public function onWorkerStart(SwServer $server, int $workerId)
    {
        $workerType = $server->taskworker ? 'task' : 'worker';

        // {{
        $this->setRunUser();
        $this->setProcessName($workerType . ' #' . $workerId);
        // }}

        // {{ log
        $this->get('logger')->info('启动 Worker 进程', [
            'id' => $workerId,
            'pid' => $server->worker_pid,
            'type' => ucfirst($workerType)
        ]);
        // }}

        if ($workerType === 'worker') {
            // {{
            $this->bootWorkerStartItems($workerId);
            // }}
            $this->callProtocolMethod('onStart', $server, $workerId);
        }

        unset($workerType);
    }
952253
public function add(Worker $worker, AbstractThread $thread)
    {
        $this->threads[$thread->getId()] = $worker->getId();
        $this->workers[$worker->getId()][$thread->getId()] = $thread;
    }
920307
public static function onStart($serv) {
		self::setProcessName(Yesf::app()->getConfig('name', Yesf::CONF_PROJECT) . ' master');
		$pidPath = rtrim(Yesf::app()->getConfig('pid', Yesf::CONF_SERVER), '/') . '/';
		try {
			file_put_contents($pidPath . Yesf::app()->getConfig('name', Yesf::CONF_PROJECT) . '_master.pid', $serv->master_pid);
			file_put_contents($pidPath . Yesf::app()->getConfig('name', Yesf::CONF_PROJECT) . '_manager.pid', $serv->manager_pid);
		} catch (\Exception $e) {
			//忽略写入错误
		}
	}
1222728
public function addUserToProject($writerId, $pid, $email, $role = 'editor', $queue = 'primary')
    {
        $job = $this->addUserToProjectAsync($writerId, $pid, $email, $role, $queue);
        if (!isset($job['url'])) {
            throw new ServerException('Create project job returned unexpected result: ' . json_encode($job, JSON_PRETTY_PRINT));
        }

        return $this->waitForJob($job['url']);
    }
888483
public static function routerBind($worker_connections, $client_connection, $cmd, $buffer)
    {
        if (!isset($client_connection->businessworker_address) || !isset($worker_connections[$client_connection->businessworker_address])) {
            $client_connection->businessworker_address = array_rand($worker_connections);
        }
        return $worker_connections[$client_connection->businessworker_address];
    }
441430
public function initAsynPools($workerId)
    {
        $this->asynPools = [];
        if ($this->config->get('redis.enable', true)) {
            $this->addAsynPool('redisPool', new RedisAsynPool($this->config, $this->config->get('redis.active')));
        }
        if ($this->config->get('mysql.enable', true)) {
            $this->addAsynPool('mysqlPool', new MysqlAsynPool($this->config, $this->config->get('mysql.active')));
        }
        if ($this->config->get('error.dingding_enable', false)) {
            $this->addAsynPool('dingdingRest', new HttpClientPool($this->config, $this->config->get('error.dingding_url')));
        }
        $this->redis_pool = $this->asynPools['redisPool'] ?? null;
        $this->mysql_pool = $this->asynPools['mysqlPool'] ?? null;
    }
380166
public function setId($id)
    {
        if ($this->status != self::STATUS_NEW) {
            throw new \RuntimeException('Cannot set worker id after worker has started working');
        }

        $this->id = $id;
    }