向 BusinessWorker 发送心跳数据，用于保持长连接

@return void
356535
public function pingBusinessWorker()
    {
        $gateway_data        = GatewayProtocol::$empty;
        $gateway_data['cmd'] = GatewayProtocol::CMD_PING;
        foreach ($this->_workerConnections as $connection) {
            $connection->send($gateway_data);
        }
    }
61095
public function save(): void
    {
        try {
            $this->em->flush();
        } catch (OptimisticLockException $e) {
            $this->messageBus->dispatch(new Message(
                $e->getMessage(),
                $e->getCode(),
                null,
                null,
                $e
            ));
        } catch (UniqueConstraintViolationException $e) {
            $this->messageBus->dispatch(new Message(
                $e->getMessage(),
                $e->getCode(),
                null,
                null,
                $e
            ));
        }
    }
877359
protected function job(): void
    {
        $limit = $this->retrieve([
            'corporation_id' => $this->getCorporationId(),
        ]);

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

        if (! property_exists($limit, 'scalar'))
            return;

        CorporationMemberLimits::firstOrNew([
            'corporation_id' => $this->getCorporationId(),
        ])->fill([
            'limit' => $limit->scalar,
        ])->save();
    }
20882
function save()
    {
        $this->_save = false;
        if ($this->_change == 0 or $this->_change == 1)
        {
            if ($this->db->insert($this->_data, $this->table) === false)
            {
                return false;
            }
            //改变状态
            $this->_change = 1;
            $this->_current_id = $this->db->lastInsertId();
        }
        elseif ($this->_change == 2)
        {
            $update = $this->_update;
            unset($update[$this->primary]);
            if ($this->db->update($this->_current_id, $update, $this->table, $this->primary) === false)
            {
                return false;
            }
        }
        $this->notify();
        return true;
    }
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;
        }
    }
1021149
public function save()
    {
        if (count($this) == 0) {
            return;
        }
        $helper = reset($this->rows);
        $qb = new QueryBuilder($helper->getTableName());
        $qb->setQueryType(QueryBuilder::QUERY_TYPE_UPDATE);
        $qb->setModelData($this);
        $qb->addWhere($helper->getPrimaryKey(), 'IN', $this->getPrimaryKeyList());
        $statement = $qb->getStatement();
        $statement->closeCursor();
    }
378384
public function doneWorking()
	{
		$this->currentJob = null;
		Resque_Stat::incr('processed');
		Resque_Stat::incr('processed:' . (string)$this);
		Resque::redis()->del('worker:' . (string)$this);
	}
358409
public function commit() {
        if ($this->dialect) {
            $dialect = $this->dialect;
            if ($dialect->inTransaction()) {
                $this->transLevel -= 1;
                if ($this->transLevel > 0) {
                    return true;
                }
                try {
                    if ($this->commitType == 'rollback') {
                        $dialect->rollBack();

                        //提交失败，因为在提交之前有人要回滚.
                        return false;
                    } else {
                        return $dialect->commit();
                    }
                } catch (\PDOException $e) {
                    $this->error = $e->getMessage();
                } finally {
                    $this->transLevel = 0;
                    $this->commitType = null;
                }
            }
        }

        return false;
    }
1010210
public function save()
    {
        $this->_dsql()->owner->beginTransaction();
        $this->hook('beforeSave');

        // decide, insert or modify
        if ($this->loaded()) {
            $res = $this->modify();
        } else {
            $res = $this->insert();
        }

        $res->hook('afterSave');
        $this->_dsql()->owner->commit();

        return $res;
    }
258031
public function setSaveHandler()
    {
        session_module_name( 'user' );
        session_set_save_handler(
            array( $this, 'open' ),
            array( $this, 'close' ),
            array( $this, 'read' ),
            array( $this, 'write' ),
            array( $this, 'destroy' ),
            array( $this, 'gc' )
            );
        return true;
    }
717576
public function handle()
    {
        $this->getStream('client')->handle(new TimeoutTimer($this->delay/4));
        $this->getStream('workerHandler')->handle(new TimeoutTimer($this->delay/4));
        $this->handleWorkerHandlerQueue();
        $this->getStream('workerHandlerStatus')->handle(new TimeoutTimer($this->delay/4));
        $this->handleExpired();
        $this->handleExpiredWorkerHandlers();
    }
84741
protected function job(): void
    {
        while (true) {

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

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

            collect($customs_offices)->each(function ($customs_office) {

                CorporationCustomsOffice::firstOrNew([
                    'corporation_id' => $this->getCorporationId(),
                    'office_id'      => $customs_office->office_id,
                ])->fill([
                    'system_id'                   => $customs_office->system_id,
                    'reinforce_exit_start'        => $customs_office->reinforce_exit_start,
                    'reinforce_exit_end'          => $customs_office->reinforce_exit_end,
                    'corporation_tax_rate'        => $customs_office->corporation_tax_rate ?? null,
                    'allow_alliance_access'       => $customs_office->allow_alliance_access,
                    'alliance_tax_rate'           => $customs_office->alliance_tax_rate ?? null,
                    'allow_access_with_standings' => $customs_office->allow_access_with_standings,
                    'standing_level'              => $customs_office->standing_level ?? null,
                    'excellent_standing_tax_rate' => $customs_office->excellent_standing_tax_rate ?? null,
                    'good_standing_tax_rate'      => $customs_office->good_standing_tax_rate ?? null,
                    'neutral_standing_tax_rate'   => $customs_office->neutral_standing_tax_rate ?? null,
                    'bad_standing_tax_rate'       => $customs_office->bad_standing_tax_rate ?? null,
                    'terrible_standing_tax_rate'  => $customs_office->terrible_standing_tax_rate ?? null,
                ])->save();

                $this->known_structures->push($customs_office->office_id);

            });

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

        // Cleanup customs offices that were not in the response.
        CorporationCustomsOffice::where('corporation_id', $this->getCorporationId())
            ->whereNotIn('office_id', $this->known_structures->flatten()->all())
            ->delete();
    }
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);
        }
    }
751796
public function handle()
    {
        if ($this->fnUpdate)
        {
            $this->worker->addFunction(
                $this->fnReload(),
                [$this, 'fnUpdate']
            );
        }

        foreach ($this->map as $name => $callable)
        {
            $this->worker->addFunction($name, $callable);
        }

        while ($this->worker->work())
        {
            continue;
        }

        return null;
    }
1108320
protected function attachUserProcesses(): void
    {
        // create Reload Worker
        if ($reloader = $this->createHotReloader()) {
            $this->server->addProcess($reloader);
        }

        foreach ($this->processCallbacks as $name => $callback) {
            $this->createProcess($name, $callback);
        }
    }
1213341
public function saveHandler()
    {
        if ($this->changed) {
            $class = $this->getRemoteClass();
            (new $class)->entity->updateRelatedThroughTable($this->table, $this->remoteAttributeName, $this->relatedAttributeName, $this->getIds(), $this->model, $this->filterList);
        }
    }
540587
public function save()
    {
        $em = Application::getFacadeApplication()->make(EntityManagerInterface::class);
        $em->persist($this);
        $em->flush();
    }