执行下个任务
@param string $connectorName
@param string $queue
@param int    $delay
@param int    $sleep
@param int    $maxTries
@return void
@throws Exception
193211
public function runNextJob($connectorName, $queue, $delay = 0, $sleep = 3, $maxTries = 0)
    {

        $job = $this->getNextJob($this->queue->driver($connectorName), $queue);

        if (!$job) {
            $this->runJob($job, $connectorName, $maxTries, $delay);
            return;
        }

        $this->sleep($sleep);
    }
1028627
public function flushQueue(string $connectionName = null)
    {
        if (empty($this->queues)) {
            return;
        }

        if ($connectionName === null) {
            foreach ($this->queues as $queueConnectionName => $queue) {
                $this->flushQueueItem($queueConnectionName, $queue);
            }
        } elseif (array_key_exists($connectionName, $this->queues)) {
            $this->flushQueueItem($connectionName, $this->queues[$connectionName]);
        }
    }
908538
public function receive($maxMessages = null, $timeout = null, ZendQueue $queue = null)
    {
        $maxMessages = $maxMessages === null ? 1 : max(1, (int) $maxMessages);
        $timeout = (int) ($timeout === null ? self::RECEIVE_TIMEOUT_DEFAULT : $timeout);
        if ($queue === null) {
            $queue = $this->_queue;
        }
        $queueId = $this->getQueueId($queue->getName());

        $msgs = [];
        $microtime = microtime(true); // cache microtime
        $microtimeInt = (int) $microtime;

        // start transaction handling
        try {
            if ($maxMessages > 0) { // ZF-7666 LIMIT 0 clause not included.
                $this->db->beginTransaction();
                $statement = $this->db->prepare("
                    select *
                    from QueueMessages
                    where queue_id = ? and (handle is null or timeout + {$timeout} < {$microtimeInt})
                    limit {$maxMessages}
                    for update
                ");
                $statement->bindValue(1, $queueId);
                $r = $statement->execute();

                foreach ($statement->fetchAll() as $data) {
                    // setup our changes to the message
                    $data['handle'] = md5(uniqid(rand(), true));

                    // update the database
                    $count = $this->db->executeUpdate(
                        "
                            update QueueMessages
                            set handle = ?, timeout = ?
                            where message_id = ? and (handle is null or timeout + {$timeout} < {$microtimeInt})
                        ",
                        [$data['handle'], $microtime, $data['message_id']]
                    );

                    // we check count to make sure no other thread has gotten
                    // the rows after our select, but before our update.
                    if ($count > 0) {
                        $msgs[] = $data;
                    }
                }
                $this->db->commit();
            }
        } catch (Exception $e) {
            $this->db->rollback();
            throw new RuntimeException($e->getMessage(), $e->getCode(), $e);
        }

        $options = [
            'queue' => $queue,
            'data' => $msgs,
            'messageClass' => $queue->getMessageClass(),
        ];

        $classname = $queue->getMessageSetClass();

        return new $classname($options);
    }
363150
public function process($connectionName, $job, WorkerOptions $options)
    {
        try {
            // First we will raise the before job event and determine if the job has already ran
            // over its maximum attempt limits, which could primarily happen when this job is
            // continually timing out and not actually throwing any exceptions from itself.
            $this->raiseBeforeJobEvent($connectionName, $job);

            $this->markJobAsFailedIfAlreadyExceedsMaxAttempts(
                $connectionName, $job, (int) $options->maxTries
            );

            // Here we will fire off the job and let it process. We will catch any exceptions so
            // they can be reported to the developers logs, etc. Once the job is finished the
            // proper events will be fired to let any listeners know this job has finished.
            $job->fire();

            $this->raiseAfterJobEvent($connectionName, $job);
        } catch (Exception $e) {
            $this->handleJobException($connectionName, $job, $options, $e);
        } catch (Throwable $e) {
            $this->handleJobException(
                $connectionName, $job, $options, new FatalThrowableError($e)
            );
        }
    }
1360684
public function pop( $connectionName, $queue = null, $delay = 0, $sleep = 3, $maxTries = 0 )
	{
		try
		{
			$connection = $this->manager->connection( $connectionName );

			$job = $this->getNextJob( $connection, $queue );

			// If we're able to pull a job off of the stack, we will process it and
			// then immediately return back out. If there is no job on the queue
			// we will "sleep" the worker for the specified number of seconds.
			if ( !is_null( $job ) )
				return $this->process( $this->manager->getName( $connectionName ), $job, $maxTries, $delay );
		}
		catch ( Exception $e )
		{
			if ( $this->exceptions )
				$this->exceptions->report( $e );
		}

		$this->sleep( $sleep );

		return ['job' => null, 'failed' => false];
	}
1097986
private function listenQueue($queueServiceName) {
        /* @var $queueService QueueInterface */
        $queueService = $this->appContext->getService($queueServiceName);
        /* @var $eventManager EventManagerInterface */
        $eventManager = $this->appContext->getService('eventManager');

        while (true) {
            while (true) {
                //出队并判断会不会是因为超时返回，如果是超时返回(返回值为null)，则继续监听队列；
                //如果不是超时而是异步事件到达(返回值不为null)，则执行其对应的事件监听器
                $event = $queueService->dequeue();
                if (is_null($event)) {
                    break;
                }

                $eventManager->trigger($event, true);

                pcntl_signal_dispatch();
            }

            pcntl_signal_dispatch();
        }
    }
111538
public function release($message, $delay = 0)
    {
        if (!$message instanceof QueueMessage) {
            $msg = new QueueMessage();
            $msg->setId($message);
        }

        $message->setDelay($delay);

        $this->driver->release($message);
    }
550814
function connect( $server, $db, $user, $password, $socketPath, $charset = null, $port = false )
    {
        $connection = false;

        if ( $socketPath !== false )
        {
            ini_set( "mysqli.default_socket", $socketPath );
        }

        if ( $this->UsePersistentConnection == true )
        {
            // Only supported on PHP 5.3 (mysqlnd)
            if ( version_compare( PHP_VERSION, '5.3' ) > 0 )
                $server = 'p:' . $server;
            else
                eZDebug::writeWarning( 'mysqli only supports persistent connections when using php 5.3 and higher', __METHOD__ );
        }

        $oldHandling = eZDebug::setHandleType( eZDebug::HANDLE_EXCEPTION );
        eZDebug::accumulatorStart( 'mysqli_connection', 'mysqli_total', 'Database connection' );
        try {
            $connection = mysqli_connect( $server, $user, $password, null, (int)$port, $socketPath );
        } catch( ErrorException $e ) {}
        eZDebug::accumulatorStop( 'mysqli_connection' );
        eZDebug::setHandleType( $oldHandling );

        $maxAttempts = $this->connectRetryCount();
        $waitTime = $this->connectRetryWaitTime();
        $numAttempts = 1;
        while ( !$connection && $numAttempts <= $maxAttempts )
        {
            sleep( $waitTime );

            $oldHandling = eZDebug::setHandleType( eZDebug::HANDLE_EXCEPTION );
            eZDebug::accumulatorStart( 'mysqli_connection', 'mysqli_total', 'Database connection' );
            try {
                $connection = mysqli_connect( $server, $user, $password, null, (int)$port, $socketPath );
            } catch( ErrorException $e ) {}
            eZDebug::accumulatorStop( 'mysqli_connection' );
            eZDebug::setHandleType( $oldHandling );

            $numAttempts++;
        }
        $this->setError();

        $this->IsConnected = true;

        if ( !$connection )
        {
            eZDebug::writeError( "Connection error: Couldn't connect to database server. Please try again later or inform the system administrator.\n{$this->ErrorMessage}", __CLASS__ );
            $this->IsConnected = false;
            throw new eZDBNoConnectionException( $server, $this->ErrorMessage, $this->ErrorNumber );
        }

        if ( $this->IsConnected && $db != null )
        {
            $ret = mysqli_select_db( $connection, $db );
            if ( !$ret )
            {
                $this->setError( $connection );
                eZDebug::writeError( "Connection error: Couldn't select the database. Please try again later or inform the system administrator.\n{$this->ErrorMessage}", __CLASS__ );
                $this->IsConnected = false;
            }
        }

        if ( $charset !== null )
        {
            $originalCharset = $charset;
            $charset = eZCharsetInfo::realCharsetCode( $charset );
        }

        if ( $this->IsConnected and $charset !== null )
        {
            $status = mysqli_set_charset( $connection, eZMySQLCharset::mapTo( $charset ) );
            if ( !$status )
            {
                $this->setError();
                eZDebug::writeWarning( "Connection warning: " . mysqli_errno( $connection ) . ": " . mysqli_error( $connection ), "eZMySQLiDB" );
            }
        }

        return $connection;
    }
836740
public function executeQueue(Queue $queue, $max, $timeout = 30)
    {
        foreach($this->receiveQueueMessages($queue, $max, $timeout) as $message) {
            try {
                $message->execute();
                $this->deleteMessage($message);
            } catch(\Exception $e) {
                $this->log($message, $e);
            }
        }
    }
92209
public function listen($connection, $queue, $delay, $memory, $timeout = 60)
    {
        $process = $this->makeProcess($connection, $queue, $delay, $memory, $timeout);

        while(true) {
            $this->runProcess($process, $memory);
        }
    }
628357
public function onMessage(Connection $conn, Message $msg)
    {
        $this->totalRdyCount--;
        $this->stats["messagesReceived"]++;

        yield $this->handleMessage($msg);
        // 收到消息后更新rdy
        // 李文告知不需要每次收到消息就调整RDY
        // $this->maybeUpdateRDY($conn);
    }
14887
public function push($job, $delay = 0, $queue = null, array $options = [])
    {
        $message = $this->getMessageByJob($job);
        $message->setDelay($delay);
        $message->setQueueName($queue);
        $message->setOptions($options);

        return $this->driver->push($message);
    }
1017683
public function add($connection, $queue, $body, $exception)
    {
        // For B/C
        $created = (new \DateTime('now'))->format('Y-m-d H:i:s');

        $sql = 'INSERT INTO ' . $this->table .
            ' (connection, queue, body, exception, created)' .
            ' VALUES (:connection, :queue, :body, :exception, :created)';

        $stat = $this->pdo->prepare($sql);
        $stat->bindValue(':connection', $connection);
        $stat->bindValue(':queue', $queue);
        $stat->bindValue(':body', $body);
        $stat->bindValue(':exception', $exception);
        $stat->bindValue(':created', $created);

        $stat->execute();

        return $this->pdo->lastInsertId();
    }
1054638
public function run($maxSeconds = null)
    {
        if (!$this->isConnected()) {
            throw new ClientException("Client has to be connected.");
        }

        $this->running = true;
        $startTime = microtime(true);
        $stopTime = null;
        if ($maxSeconds !== null) {
            $stopTime = $startTime + $maxSeconds;
        }

        do {
            if (!empty($this->queue)) {
                $frame = array_shift($this->queue);

            } else {
                if (($frame = $this->reader->consumeFrame($this->readBuffer)) === null) {
                    $now = microtime(true);
                    $nextStreamSelectTimeout = $nextHeartbeat = ($this->lastWrite ?: $now) + $this->options["heartbeat"];
                    if ($stopTime !== null && $stopTime < $nextStreamSelectTimeout) {
                        $nextStreamSelectTimeout = $stopTime;
                    }
                    $tvSec = max(intval($nextStreamSelectTimeout - $now), 0);
                    $tvUsec = max(intval(($nextStreamSelectTimeout - $now - $tvSec) * 1000000), 0);

                    $r = [$this->getStream()];
                    $w = null;
                    $e = null;

                    if (($n = @stream_select($r, $w, $e, $tvSec, $tvUsec)) === false) {
                        $lastError = error_get_last();
                        if ($lastError !== null &&
                            preg_match("/^stream_select\\(\\): unable to select \\[(\\d+)\\]:/", $lastError["message"], $m) &&
                            intval($m[1]) === PCNTL_EINTR
                        ) {
                            // got interrupted by signal, dispatch signals & continue
                            pcntl_signal_dispatch();
                            $n = 0;

                        } else {
                            throw new ClientException(sprintf(
                                "stream_select() failed: %s",
                                $lastError ? $lastError["message"] : "Unknown error."
                            ));
                        }
                    }

                    $now = microtime(true);

                    if ($now >= $nextHeartbeat) {
                        $this->writer->appendFrame(new HeartbeatFrame(), $this->writeBuffer);
                        $this->flushWriteBuffer();
                    }

                    if ($stopTime !== null && $now >= $stopTime) {
                        break;
                    }

                    if ($n > 0) {
                        $this->feedReadBuffer();
                    }

                    continue;
                }
            }

            /** @var AbstractFrame $frame */

            if ($frame->channel === 0) {
                $this->onFrameReceived($frame);

            } else {
                if (!isset($this->channels[$frame->channel])) {
                    throw new ClientException(
                        "Received frame #{$frame->type} on closed channel #{$frame->channel}."
                    );
                }

                $this->channels[$frame->channel]->onFrameReceived($frame);
            }


        } while ($this->running);
    }
1267275
public static function do(ConnectorInterface $connector, $voter, $publicWif, $author, $permlink, $weight)
    {
        $chainName = $connector->getPlatform();
        /** @var CommandQueryData $tx */
        $tx = Transaction::init($connector);
        $tx->setParamByKey(
            '0:operations:0',
            [
                'vote',
                [
                    'voter'    => $voter,
                    'author'   => $author,
                    'permlink' => $permlink,
                    'weight'   => $weight
                ]
            ]
        );

        $command = new BroadcastTransactionCommand($connector);////        echo '<pre>' . var_dump($commandQueryData->getParams(), $properties2) . '<pre>'; die; //FIXME delete it
        Transaction::sign($chainName, $tx, ['posting' => $publicWif]);

        $answer = $command->execute(
            $tx
        );

        return $answer;
    }
1312369
public function execute($i, $maxProcesses, $suite, $currentProcessCounter, $isFirstOnItsThread = false)
    {
        return array_change_key_case(static::cleanEnvVariables($_SERVER) + $_ENV + [
            self::ENV_TEST_CHANNEL => (int) $i,
            self::ENV_TEST_CHANNEL_READABLE => 'test_'.$i,
            self::ENV_TEST_CHANNELS_NUMBER => (int) $maxProcesses,
            self::ENV_TEST_ARGUMENT => (string) $suite,
            self::ENV_TEST_INCREMENTAL_NUMBER => (int) $currentProcessCounter,
            self::ENV_TEST_IS_FIRST_ON_CHANNEL => (int) $isFirstOnItsThread,
        ], CASE_UPPER);
    }
124539
public function later($delay, Queue $queue)
    {
        $connection = property_exists($this, 'connection') ? $this->connection : null;

        $queueName = property_exists($this, 'queue') ? $this->queue : null;

        return $queue->connection($connection)->laterOn(
            $queueName ?: null, $delay, new SendQueuedMailable($this)
        );
    }