pop

@param string $queue

@return QueueMessage
@throws \Exception
@throws \InvalidArgumentException
@throws \Throwable
485161
public function pop($queue = null)
    {
        $queue = $queue ?: $this->queue;

        $now = new \DateTimeImmutable('now');

        $sql = 'SELECT * FROM ' . $this->table .
            ' WHERE queue = :queue AND visibility < :visibility' .
            ' AND (reserved IS NULL OR reserved < :reserved)' .
            ' FOR UPDATE';

        $this->pdo->beginTransaction();

        $stat = $this->pdo->prepare($sql);
        $stat->bindValue(':queue', $queue);
        $stat->bindValue(':visibility', $now->format('Y-m-d H:i:s'), \PDO::PARAM_STR);
        $stat->bindValue(
            ':reserved',
            $now->modify('-' . $this->timeout . 'seconds')->format('Y-m-d H:i:s'),
            \PDO::PARAM_STR
        );

        try {
            $stat->execute();

            $data = $stat->fetch(\PDO::FETCH_ASSOC);

            if (!$data) {
                $this->pdo->commit();

                return null;
            }

            $data['attempts']++;

            $sql = 'UPDATE ' . $this->table . ' SET reserved = :reserved, attempts = :attempts WHERE id = :id';

            $stat = $this->pdo->prepare($sql);
            $stat->bindValue(':reserved', $now->format('Y-m-d H:i:s'));
            $stat->bindValue(':attempts', $data['attempts'] + 1);
            $stat->bindValue(':id', $data['id']);

            $stat->execute();

            $this->pdo->commit();
        } catch (\Throwable $t) {
            $this->pdo->rollBack();
            throw $t;
        }

        $message = new QueueMessage();

        $message->setId($data['id']);
        $message->setAttempts($data['attempts']);
        $message->setBody(json_decode($data['body'], true));
        $message->setRawBody($data['body']);
        $message->setQueueName($queue);

        return $message;
    }
1239456
protected function getNextMessage($queue)
    {
        $queue = (array) $queue;

        foreach ($queue as $queueName) {
            if ($message = $this->manager->pop($queueName)) {
                return $message;
            }
        }

        return null;
    }
583541
public function get(string $queue = 'default')
    {
        $message = yield $this->kafka->consume();
        if (!is_object($message)) {
            return false;
        }

        if ($message->err === RD_KAFKA_RESP_ERR_NO_ERROR) {//正常
            return $message->payload;
        } elseif ($message->err === RD_KAFKA_RESP_ERR__PARTITION_EOF) {//暂时没有消息
            return null;
        } elseif ($message->err === RD_KAFKA_RESP_ERR__TIMED_OUT) {//超时
            return false;
        } else {
            throw new \RdKafka\Exception('RD_KAFKA_RESP_ERR', $message->err);//其他异常
        }
    }
265556
def pop(self, queue_name):
        
        queue = self._get_queue(queue_name)
        message = queue.read()

        if message:
            data = message.get_body()
            return data
719823
public function pop($queue = null)
    {
        $queue = $this->getQueue($queue);

        $this->database->trans_begin();

        if ($job = $this->getNextAvailableJob($queue)) {
            $job = $this->markJobAsReserved($job);

            $this->database->trans_commit();

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

        $this->database->trans_commit();
    }
925505
public function delete(QueueMessage $message)
    {
        $queue = $message->getQueueName() ?: $this->queue;

        $this->client->deleteMessage($queue, $message->getId(), $message->get('reservation_id'));

        return $this;
    }
199422
protected function retrieveNextJob($queue)
    {
        if (! is_null($this->blockFor)) {
            return $this->blockingPop($queue);
        }

        return $this->getConnection()->eval(
            LuaScripts::pop(), 2, $queue, $queue.':reserved',
            $this->availableAt($this->retryAfter)
        );
    }
759385
public function pop($queue = null)
	{
		$queue = $this->getQueue($queue);

		$job = $this->iron->getMessage($queue);

		// If we were able to pop a message off of the queue, we will need to decrypt
		// the message body, as all Iron.io messages are encrypted, since the push
		// queues will be a security hazard to unsuspecting developers using it.
		if ( ! is_null($job))
		{
			$job->body = $this->crypt->decrypt($job->body);

			return new IronJob($this->container, $this, $job);
		}
	}
1009915
public function pop($topic, $timeout = PHP_INT_MAX)
    {
        if (($msg = $this->do_pop($topic, $timeout)) !== false) {
            $this->fireEvent('messageQueue:pop', ['topic' => $topic, 'msg' => $msg]);
        }

        return $msg;
    }
1093429
public function queuePurge($queue = '', $nowait = false)
    {
        return $this->getClient()->queuePurge($this->getChannelId(), $queue, $nowait);
    }
133083
protected function retrieveNextJob($queue)
    {
        if (!is_null($this->blockFor)) {
            return $this->blockingPop($queue);
        }

        $job      = $this->redis->lpop($queue);
        $reserved = false;

        if ($job) {
            $reserved = json_decode($job);
            $reserved['attempts']++;
            $reserved = json_encode($reserved);
            $this->redis->zAdd($queue . ':reserved', $this->availableAt($this->retryAfter), $reserved);
        }

        return [$job, $reserved];
    }
402682
function pop()
    {
        $ret = \Swoole::$php->redis($this->redis_factory_key)->lPop($this->key);
        if ($ret)
        {
            return unserialize($ret);
        }
        else
        {
            return false;
        }
    }
320161
public function pop($queue = null)
    {
        $this->migrate($prefixed = $this->getQueue($queue));

        if (empty($nextJob = $this->retrieveNextJob($prefixed))) {
            return;
        }

        [$job, $reserved] = $nextJob;

        if ($reserved) {
            return new RedisJob(
                $this->container, $this, $job,
                $reserved, $this->connectionName, $queue ?: $this->default
            );
        }
    }
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;
    }
351560
private function doPopMessage($queueName)
    {
        $query = 'SELECT id, message FROM bernard_messages
                  WHERE queue = :queue AND visible = :visible
                  ORDER BY sentAt LIMIT 1 '.$this->connection->getDatabasePlatform()->getForUpdateSql();

        list($id, $message) = $this->connection->fetchArray($query, [
            'queue' => $queueName,
            'visible' => true,
        ]);

        if ($id) {
            $this->connection->update('bernard_messages', ['visible' => 0], compact('id'));

            return [$message, $id];
        }
    }
309242
public static function loadPayload($queue, $payload)
    {
        $payload = json_decode($payload, true);

        if (!is_array($payload) or !count($payload)) {
            throw new \InvalidArgumentException('Supplied $payload must be a json encoded array.');
        }

        return new static($queue, $payload['id'], $payload['class'], $payload['data']);
    }
388309
def pop(non_block=false)
      handle_interrupt do
        @mutex.synchronize do
          while true
            if @que.empty?
              if non_block
                raise ThreadError, "queue empty"
              else
                begin
                  @num_waiting += 1
                  @cond.wait @mutex
                ensure
                  @num_waiting -= 1
                end
              end
            else
              return @que.shift
            end
          end
        end
      end
    end