获取下个有效任务

@param string|null $queue
@return StdClass|null
320017
protected function getNextAvailableJob($queue)
    {

        $job = $this->db->name($this->table)
            ->lock(true)
            ->where('queue', $this->getQueue($queue))
            ->where(function (Query $query) {
                $query->where(function (Query $query) {
                    $query->whereNull('reserved_at')
                        ->where('available_at', '<=', $this->currentTime());
                });

                //超时任务重试
                $expiration = Carbon::now()->subSeconds($this->retryAfter)->getTimestamp();

                $query->whereOr(function (Query $query) use ($expiration) {
                    $query->where('reserved_at', '<=', $expiration);
                });
            })
            ->order('id', 'asc')
            ->find();

        return $job ? (object) $job : null;
    }
1043203
public function size($queue = null)
    {
        $queue = $this->getQueue($queue);

        return (int) $this->pheanstalk->statsTube($queue)->current_jobs_ready;
    }
1333186
public static function daemonStart(&$queue){
        $recordHandler = self::getRedisForStatics();

        $worker_pid = self::getPid();
        $key        = self::worker_status_monitoring_key_prefix . $worker_pid;
        $value      = json_encode([
            'queue'   => $queue,
            'comment' => 'worker_daemon_start',
            'ts'      => time(),
        ]);
        $recordHandler->lPush($key, $value);                // 存到 list 的头部
        unset($recordHandler);
    }
913486
protected function getNextJob( $connection, $queue )
	{
		if ( is_null( $queue ) )
			return $connection->pop();

		foreach ( explode( ',', $queue ) as $queue )
			if ( !is_null( $job = $connection->pop( $queue ) ) )
				return $job;

		return null;
	}
662215
public function getMessagesCurrentCountOnQueue($queue)
    {
        try {
            return $this->queue->getQueue($queue)->size;
        } catch (\Http_Exception $ex) {
            \PHPUnit\Framework\Assert::fail("queue [$queue] not found");
        }
    }
328016
public function reserve(string $queue): ?Job
    {
        $item = $this->client->lpop(sprintf('queue:%s', $queue));

        if (!$item) {
            return null;
        }

        $payload = json_decode($item, true);

        $id = $payload['id'] ?? null;

        return new Job($payload['class'], $queue, $payload['args'], $id);
    }
216350
protected function setQueue(Queue $queue)
    {
        $this->queue = $queue;
        $this->queue->setProducer($this);

        return $this;
    }
634667
public function pop($queue)
    {
        $item = $this->getClient()->lpop($this->getKey(self::QUEUE_KEY . $queue));

        if (!$item) {
            return null;
        }

        return json_decode($item, true);
    }
58425
private function _ensureExchangeForQueue($queue)
    {
        $serverId = $this->_getQueueServerId($queue);
        $server   = &$this->servers[$serverId];
        
        $reason       = self::REASON_MAIN;
        $exchangeName = self::DEFAULT_EXCHANGE;
        
        if (isset($server[self::SERVER_EXCHANGES][$reason][$exchangeName])) {
            return $server[self::SERVER_EXCHANGES][$reason][$exchangeName];
        }
        
        $this->_serverEnsureConnection($server, $reason);
        $this->_serverEnsureChannel($server, self::REASON_MAIN);
        $exchangeObj = $this->_setupExchange($server, self::REASON_MAIN, $exchangeName);
        
        $server[self::SERVER_EXCHANGES][$reason][$exchangeName] = $exchangeObj;
        
        return $exchangeObj;
    }
354095
public function execute($queue)
    {
        if (($user = $this->getUser()) != null) {
            $originalImage = Yii::getAlias('@runtime/') . $user->id . '_avatar.jpg';
            //下载图片
            if (($image = @file_get_contents($this->faceUrl)) != false) {
                //保存原图
                file_put_contents($originalImage, $image);
                AvatarHelper::save($user, $originalImage);
            };
        }
    }
410098
public function getQueue($queuePath)
    {
        $httpCallContext = new HttpCallContext();
        $httpCallContext->setPath($queuePath);
        $httpCallContext->setMethod(Resources::HTTP_GET);
        $httpCallContext->addStatusCode(Resources::STATUS_OK);
        $response = $this->sendHttpContext($httpCallContext);
        $queueInfo = new QueueInfo();
        $queueInfo->parseXml($response->getBody());

        return $queueInfo;
    }
1281417
protected function getNextAvailableJob($queue)
    {
        $expiration = Carbon::now()->subSeconds($this->retryAfter)->getTimestamp();
        $sql = 'SELECT * FROM ' . $this->table . ' WHERE queue =\'' . $this->getQueue($queue) . '\' 
                AND (
                        (reserved_at is NULL AND available_at <=' . $this->currentTime() . ')
                        OR 
                        (reserved_at <= ' . $expiration . ')
                    ) 
                order by id asc limit 1 ';

        $job = $this->database->createCommand($sql)->queryOne();

        return $job ? new DatabaseJobRecord((object)$job) : null;
    }
344101
public function _countMessage($queue)
    {
        list($queue, $messageCount) = $this->getChannel()->queue_declare($queue, true);
        return $messageCount;
    }
617041
protected function getNextJob($connection, $queue)
    {
        try {
            if (is_null($queue)) {
                return $connection->pop();
            }

            foreach (explode(',', $queue) as $queue) {
                if (! is_null($job = $connection->pop($queue))) {
                    return $job;
                }
            }
        }
        catch (Exception $e) {
            $this->handleException($e);
        }
        catch (Throwable $e) {
            $this->handleException(new FatalThrowableError($e));
        }
    }
348846
public function get($queue = '', $noAck = false)
    {
        return $this->getClient()->get($this->getChannelId(), $queue, $noAck);
    }
468401
function getQueue($token)
	{
		$a = func_get_args();
		Debugger::log(__CLASS__ . ": " . __METHOD__ . "; args: " . print_r($a, true));

		$q = new Queue();
		$q->setQueueID($token);
		$q->setQueuesModel($this);
		return $q;
	}
138099
public function pop($queue = null)
    {
        $queue = $this->getQueue($queue);

        return $this->database->transaction(function () use ($queue)
        {
            if (! is_null($job = $this->getNextAvailableJob($queue))) {
                $this->markJobAsReserved($job->id, $job->attempts);

                return new DatabaseJob(
                    $this->container, $this, $job, $queue
                );
            }
        });
    }