初始化所有的worker实例，主要工作为获得格式化所需数据及监听端口
@return void
345044
protected static function initWorkers()
    {
        foreach(static::$_workers as $worker)
        {
            // 没有设置worker名称，则使用none代替
            if(empty($worker->name))
            {
                $worker->name = 'none';
            }
            // 获得所有worker名称中最大长度
            $worker_name_length = strlen($worker->name);
            if(static::$_maxWorkerNameLength < $worker_name_length)
            {
                static::$_maxWorkerNameLength = $worker_name_length;
            }
            // 获得所有_socketName中最大长度
            $socket_name_length = strlen($worker->getSocketName());
            if(static::$_maxSocketNameLength < $socket_name_length)
            {
                static::$_maxSocketNameLength = $socket_name_length;
            }
            $user_name_length = strlen($worker->user);
            if(static::$_maxUserNameLength < $user_name_length)
            {
                static::$_maxUserNameLength = $user_name_length;
            }
        }
    }
1272029
protected function job(): void
    {
        $balances = $this->retrieve([
            'corporation_id' => $this->getCorporationId(),
        ]);

        if ($balances->isCachedLoad()) return;

        collect($balances)->each(function ($balance) {

            CorporationWalletBalance::firstOrNew([
                'corporation_id' => $this->getCorporationId(),
                'division'       => $balance->division,
            ])->fill([
                'balance' => $balance->balance,
            ])->save();

        });
    }
895736
protected function listenForEvents()
    {
        // -- Worker处理的事件
        //listener:beforeDaemonLoop
        //listener:beforeGetJob
        //listener:afterGetJob
        //listener:beforeJobHandle
        //listener:logger
        //listener:afterJobHandle
        //listener:jobException
        //listener:jobFailed
        //listener:afterDaemonLoop
        //listener:stop
        $this->eventsManager->attach('listener', function (Event $event, $source, $data) {
            $date = date('Y-m-d H:i:s');
            $class = get_class($source);
            $event = $event->getType();
            $payload = '';
            $message = '';
            if (is_array($data)) {
                $job = $data[0];
                $e = $data[1];
                if ($job instanceof Job) {
                    $payload = $job->getPayload();
                } elseif (is_scalar($data)) {
                    $payload = $data;
                } else {
                    $payload = serialize($job);
                }
                if ($e instanceof \Exception) {
                    $message = $data[1]->getMessage();
                } else {
                    $message = serialize($job);
                }
            } else {
                if ($data instanceof Job) {
                    $payload = $data->getPayload();
                } elseif (is_scalar($data)) {
                    $message = $data;
                } else {
                    $message = serialize($data);
                }
            }
            $this->line('[' . $date . '] Event=listener:' . $event . ', Class=' . $class . ', Msg=' . $message . ', Payload=' . $payload);
        });
    }
654364
public function onWorkerStart()
    {
        if (empty($this->serverRoot))
        {
            throw new Exception('server root not set, please use WebServer::addRoot($domain, $root_path) to set server root path');
        }

        // 初始化HttpCache
        HttpCache::init();
        session_name(self::$sessionName);
        HttpCache::$sessionName = self::$sessionName;

        // 尝试执行开发者设定的onWorkerStart回调
        if ($this->_onWorkerStart)
        {
            call_user_func($this->_onWorkerStart, $this);
        }
    }
621437
public function index()
    {
        //  Command line only
        $oInput = Factory::service('Input');
        if (Environment::is(Environment::ENV_PROD) && !$oInput::isCli()) {
            header($oInput->server('SERVER_PROTOCOL') . ' 401 Unauthorized');
            echo '<h1>' . lang('unauthorised') . '</h1>';
            exit(401);
        }

        // --------------------------------------------------------------------------

        /**
         * Look for a controller, app version first then the module's cron controllers directory
         */
        $aControllerPaths = [
            NAILS_APP_PATH . 'application/modules/cron/controllers/',
        ];

        $nailsModules = Components::modules();

        foreach ($nailsModules as $oModule) {

            if ($oModule->moduleName === $this->sModuleName) {
                $aControllerPaths[] = $oModule->path . 'cron/controllers/';
                break;
            }
        }

        //  Look for a valid controller
        $sControllerName    = ucfirst($this->sClassName) . '.php';
        $bDidfindController = false;

        foreach ($aControllerPaths as $sPath) {

            $sControllerPath = $sPath . $sControllerName;

            if (is_file($sControllerPath)) {
                $bDidfindController = true;
                break;
            }
        }

        if (!empty($bDidfindController)) {

            //  Load the file and try and execute the method
            require_once $sControllerPath;

            $this->sModuleName = 'Nails\\Cron\\' . ucfirst($this->sModuleName) . '\\' . ucfirst($this->sClassName);

            if (class_exists($this->sModuleName)) {

                //  New instance of the controller
                $oInstance = new $this->sModuleName($this);

                if (is_callable([$oInstance, $this->sMethod])) {

                    //  Begin timing
                    $this->writeLog('Starting job');

                    if (!empty($this->aParams)) {

                        $this->writeLog('Passed parameters');
                        $this->writeLog(print_r($this->aParams, true));
                    }

                    $iStart = microtime(true) * 10000;

                    call_user_func_array([$oInstance, $this->sMethod], $this->aParams);

                    $iEnd      = microtime(true) * 10000;
                    $iDuration = ($iEnd - $iStart) / 10000;
                    $this->writeLog('Finished job');
                    $this->writeLog('Job took ' . number_format($iDuration, 5) . ' seconds');

                } else {

                    $this->writeLog('Cannot call method "' . $this->sMethod . '"');
                }

            } else {

                $this->writeLog(
                    '"' . $this->sModuleName . '" is incorrectly configured; could not find class ' . $this->sModuleName . '  at path ' . $sControllerPath
                );
            }

        } else {

            $this->writeLog(
                '"' . $this->sModuleName . '/' . $this->sClassName . '/' . $this->sMethod . '" is not a valid route.'
            );
        }
    }
278953
protected function onWorkerStart()
    {
        if (!class_exists('\Protocols\GatewayProtocol')) {
            class_alias('GatewayWorker\Protocols\GatewayProtocol', 'Protocols\GatewayProtocol');
        }

        if (!is_array($this->registerAddress)) {
            $this->registerAddress = array($this->registerAddress);
        }
        $this->connectToRegister();

        \GatewayWorker\Lib\Gateway::setBusinessWorker($this);
        \GatewayWorker\Lib\Gateway::$secretKey = $this->secretKey;
        if ($this->_onWorkerStart) {
            call_user_func($this->_onWorkerStart, $this);
        }
        
        if (is_callable($this->eventHandler . '::onWorkerStart')) {
            call_user_func($this->eventHandler . '::onWorkerStart', $this);
        }

        if (function_exists('pcntl_signal')) {
            // 业务超时信号处理
            pcntl_signal(SIGALRM, array($this, 'timeoutHandler'), false);
        } else {
            $this->processTimeout = 0;
        }

        // 设置回调
        if (is_callable($this->eventHandler . '::onConnect')) {
            $this->_eventOnConnect = $this->eventHandler . '::onConnect';
        }

        if (is_callable($this->eventHandler . '::onMessage')) {
            $this->_eventOnMessage = $this->eventHandler . '::onMessage';
        } else {
            echo "Waring: {$this->eventHandler}::onMessage is not callable\n";
        }

        if (is_callable($this->eventHandler . '::onClose')) {
            $this->_eventOnClose = $this->eventHandler . '::onClose';
        }

        if (is_callable($this->eventHandler . '::onWebSocketConnect')) {
            $this->_eventOnWebSocketConnect = $this->eventHandler . '::onWebSocketConnect';
        }

    }
28083
public function workers()
    {
        $this->sendCommand('workers');
        $res = $this->recvCommand();
        $workers = array();
        $tmp = explode("\n", $res);
        foreach ($tmp as $t) {
            if (!Connection::stringLength($t)) {
                continue;
            }

            $info = explode(' : ', $t);
            list($fd, $ip, $id) = explode(' ', $info[0]);

            $abilities = isset($info[1]) ? trim($info[1]) : '';

            $workers[] = array(
                'fd' => $fd,
                'ip' => $ip,
                'id' => $id,
                'abilities' => empty($abilities) ? array() : explode(' ', $abilities),
            );
        }

        return $workers;
    }
1339318
protected function job(): void
    {
        $unfinished_auctions = CorporationContract::join('contract_details',
            'corporation_contracts.contract_id', '=',
            'contract_details.contract_id')
            ->where('corporation_id', $this->getCorporationId())
            ->where('type', 'auction')
            ->whereNotIn('status', ['finished', 'deleted'])
            ->pluck('corporation_contracts.contract_id');

        $unfinished_auctions->each(function ($contract_id) {

            while (true) {

                try {
                    $bids = $this->retrieve([
                        'corporation_id' => $this->getCorporationId(),
                        'contract_id'    => $contract_id,
                    ]);

                    if ($bids->isCachedLoad()) return;

                    collect($bids)->each(function ($bid) use ($contract_id) {

                        ContractBid::firstOrCreate([
                            'bid_id' => $bid->bid_id,
                        ], [
                            'contract_id' => $contract_id,
                            'bidder_id' => $bid->bidder_id,
                            'date_bid' => carbon($bid->date_bid),
                            'amount' => $bid->amount,
                        ]);
                    });

                    if (! $this->nextPage($bids->pages))
                        break;

                } catch (RequestFailedException $e) {
                    if (strtolower($e->getError()) == 'contract not found') {
                        ContractDetail::where('contract_id', $contract_id)
                            ->update([
                                'status' => 'deleted',
                            ]);

                        break;
                    }

                    throw $e;
                }
            }

            // reset the page back to page one for the next contract
            $this->page = 1;
        });
    }
668520
protected function doExecute()
    {
        $queues     = $this->io->getArguments();
        $options    = $this->getWorkOptions();
        $connection = $this->getOption('connection', null) ?: $this->console->get('queue.connection');

        // Set connection as default
        if ($connection !== null) {
            $this->console->set('queue.connection', $connection);
        }

        /** @var Worker $worker */
        $worker = $this->console->container->get('queue.worker');

        // Default Queues
        if (!count($queues)) {
            $queues = $this->console->container->get('queue.manager')
                ->getConnectionConfig($connection)
                ->get('queue');
        }

        $this->listenToWorker($worker);

        if ($this->getOption('once')) {
            $worker->runNextJob($queues, $options);
        } else {
            $worker->loop($queues, $options);
        }

        return true;
    }
1251291
private void startWorkers() {
        m_logger.info("Starting {} workers", m_config.workers);
        for (int workerNo = 1; workerNo <= m_config.workers; workerNo++) {
            LoadWorker loadWorker = new LoadWorker(workerNo);
            loadWorker.start();
            m_workerList.add(loadWorker);
        }
    }
218914
protected function job(): void
    {
        $labels = $this->retrieve([
            'corporation_id' => $this->getCorporationId(),
        ]);

        if ($labels->isCachedLoad()) return;

        collect($labels)->each(function ($label) {

            CorporationContactLabel::firstOrNew([
                'corporation_id' => $this->getCorporationId(),
                'label_id'       => $label->label_id,
            ])->fill([
                'label_name' => $label->label_name,
            ])->save();
        });

        CorporationContactLabel::where('corporation_id', $this->getCorporationId())
            ->whereNotIn('label_id', collect($labels)->pluck('label_id')->flatten()->all())
            ->delete();
    }
320095
public function run()
    {
        $this->setupSignalHandlers();

        $procs = [];
        while (true) {
            $jobs = $this->runner->getJobs();
            foreach ($jobs as $job) {
                if (! array_key_exists($job->identifier(), $procs)) {
                    $procs[$job->identifier()] = [];
                }
                while (count($procs[$job->identifier()]) > $job->numWorkers()) {
                    // Stopping an excessive child.
                    echo 'Stopping an excessive child.' . PHP_EOL;
                    $proc = array_pop($procs[$job->identifier()]);
                    $status = proc_get_status($proc);
                    // Keep sending SIGTERM until the child exits.
                    while ($status['running'] === true) {
                        @proc_terminate($proc);
                        usleep(50000);
                        $status = proc_get_status($proc);
                    }
                    @proc_close($proc);
                }
                for ($i = 0; $i < $job->numWorkers(); $i++) {
                    $needStart = false;
                    if (array_key_exists($i, $procs[$job->identifier()])) {
                        $status = proc_get_status($procs[$job->identifier()][$i]);
                        if ($status['running'] !== true) {
                            $needStart = true;
                        }
                    } else {
                        $needStart = true;
                    }
                    if ($needStart) {
                        echo 'Starting a child.' . PHP_EOL;
                        $procs[$job->identifier()][$i] = proc_open(
                            sprintf('%s %d', $this->command, $job->id()),
                            $this->descriptorSpec,
                            $pipes
                        );
                    }
                }
            }
            usleep(1000000); // Reload the config after 1 second
            pcntl_signal_dispatch();
            if ($this->shutdown) {
                echo 'Shutting down, waiting for the children' . PHP_EOL;
                foreach ($procs as $k => $v) {
                    foreach ($v as $proc) {
                        $status = proc_get_status($proc);
                        // Keep sending SIGTERM until the child exits.
                        while ($status['running'] === true) {
                            @proc_terminate($proc);
                            usleep(50000);
                            $status = proc_get_status($proc);
                        }
                        @proc_close($proc);
                    }
                }
                echo 'BatchDaemon exiting' . PHP_EOL;
                exit;
            }
            // Reload the config
            $this->runner->loadConfig();
        }
    }
791217
private function start() {
        if (file_exists($this->pidFile)) {
            _throw('事件监听守护进程已经在运行中');
        }

        $queueServices = $this->extraQueueServices();

        //如果没有(需要)队列化的异步事件
        if (empty($queueServices)) {
            _throw('没有需要队列化的异步事件');
        }

        //当前进程转为守护进程
        $this->daemon();

        //为每个事件队列分别开启一个工作进程
        foreach ($queueServices as $queueServiceName) {
            $this->startWorker($queueServiceName);
        }

        //主进程做一些管理工作
        while (true) {
            //如果收到SIGTERM信号要求退出，且工作进程已经全部退出，则删除PID文件，然后退出
            if ($this->shouldExit && empty($this->workerMap)) {
                $this->log('[master] 工作进程已经全部退出，结束运行，进程ID：' . getmypid());
                unlink($this->pidFile);
                exit(0);
            }

            pcntl_signal_dispatch();
            $this->waitWorker();
            sleep(3);
        }
    }
858506
protected function job(): void
    {
        while (true) {

            $industry_jobs = $this->retrieve([
                'corporation_id' => $this->getCorporationId(),
            ]);

            if ($industry_jobs->isCachedLoad()) return;

            collect($industry_jobs)->each(function ($job) {

                CorporationIndustryJob::firstOrNew([
                    'corporation_id' => $this->getCorporationId(),
                    'job_id'         => $job->job_id,
                ])->fill([
                    'installer_id'           => $job->installer_id,
                    'facility_id'            => $job->facility_id,
                    'location_id'            => $job->location_id,
                    'activity_id'            => $job->activity_id,
                    'blueprint_id'           => $job->blueprint_id,
                    'blueprint_type_id'      => $job->blueprint_type_id,
                    'blueprint_location_id'  => $job->blueprint_location_id,
                    'output_location_id'     => $job->output_location_id,
                    'runs'                   => $job->runs,
                    'cost'                   => $job->cost ?? null,
                    'licensed_runs'          => $job->licensed_runs ?? null,
                    'probability'            => $job->probability ?? null,
                    'product_type_id'        => $job->product_type_id ?? null,
                    'status'                 => $job->status,
                    'duration'               => $job->duration,
                    'start_date'             => carbon($job->start_date),
                    'end_date'               => carbon($job->end_date),
                    'pause_date'             => property_exists($job, 'pause_date') ?
                        carbon($job->pause_date) : null,
                    'completed_date'         => property_exists($job, 'completed_date') ?
                        carbon($job->completed_date) : null,
                    'completed_character_id' => $job->completed_character_id ?? null,
                    'successful_runs'        => $job->successful_runs ?? null,
                ])->save();
            });

            if (! $this->nextPage($industry_jobs->pages))
                return;
        }
    }
286184
protected function loop(): void
    {
        $job = $this->getJob();

        if (!$job) {
            $this->sleep();

            return;
        }

        $this->log->info(sprintf('got Job %s', $job->getId()));

        $this->getEventManager()->trigger('beforeFork', $job);
        $this->setWorkingOn($job);

        $this->child = $this->fork();

        // We are the Child Process
        if ($this->child === 0 || $this->child === false) {
            $this->runAsChild($job);
        }

        // We are the parent Process
        if ($this->child > 0) {
            $this->runAsParent($job);
        }

        $this->child = null;
        $this->finishedWorking();
    }
794392
public function startWorker()
    {
        $this->pheanstalk->watch($this->queue);
        $this->pheanstalk->ignore('default');
        $buildStore = Factory::getStore('Build');

        while ($this->run) {
            // Get a job from the queue:
            $job = $this->pheanstalk->reserve();

            $this->checkJobLimit();

            // Get the job data and run the job:
            $jobData = json_decode($job->getData(), true);

            if (!$this->verifyJob($job, $jobData)) {
                continue;
            }

            $this->logger->addInfo('Received build #'.$jobData['build_id'].' from Beanstalkd');

            // If the job comes with config data, reset our config and database connections
            // and then make sure we kill the worker afterwards:
            if (!empty($jobData['config'])) {
                $this->logger->addDebug('Using job-specific config.');
                $currentConfig = Config::getInstance()->getArray();
                $config = new Config($jobData['config']);
                Database::reset($config);
            }

            try {
                $build = BuildFactory::getBuildById($jobData['build_id']);
            } catch (\Exception $ex) {
                $this->logger->addWarning('Build #' . $jobData['build_id'] . ' does not exist in the database.');
                $this->pheanstalk->delete($job);
            }

            try {
                // Logging relevant to this build should be stored
                // against the build itself.
                $buildDbLog = new BuildDBLogHandler($build, Logger::INFO);
                $this->logger->pushHandler($buildDbLog);

                $builder = new Builder($build, $this->logger);
                $builder->execute();

                // After execution we no longer want to record the information
                // back to this specific build so the handler should be removed.
                $this->logger->popHandler($buildDbLog);
            } catch (\PDOException $ex) {
                // If we've caught a PDO Exception, it is probably not the fault of the build, but of a failed
                // connection or similar. Release the job and kill the worker.
                $this->run = false;
                $this->pheanstalk->release($job);
            } catch (\Exception $ex) {
                $build->setStatus(Build::STATUS_FAILED);
                $build->setFinished(new \DateTime());
                $build->setLog($build->getLog() . PHP_EOL . PHP_EOL . $ex->getMessage());
                $buildStore->save($build);
                $build->sendStatusPostback();
            }

            // Reset the config back to how it was prior to running this job:
            if (!empty($currentConfig)) {
                $config = new Config($currentConfig);
                Database::reset($config);
            }

            // Delete the job when we're done:
            $this->pheanstalk->delete($job);
        }
    }
676617
public function restart()
    {
        $workersArgs = $this->ResqueStatus->getWorkers();

        $this->params['all'] = true;
        $this->stop();

        $this->out('<info>' . __d('cake_resque', 'Restarting workers') . '</info>');
        if (!empty($workersArgs)) {
            $debug = $this->params['debug'];
            $this->debug(__d('cake_resque', 'Found ' . count($workersArgs) . ' workers to restart'));

            foreach ($workersArgs as $workerArgs) {
                $workerArgs['debug'] = $debug;
                if (isset($workerArgs['type']) && $workerArgs['type'] === 'scheduler') {
                    $this->startScheduler($workerArgs);
                } else {
                    $this->start($workerArgs);
                }
            }
            $this->out('');
        } else {
            $this->out('<warning>' . __d('cake_resque', 'No active workers found, will start brand new worker') . '</warning>');
            $this->start();
        }
    }