出队
@param string $queue
@param bool $isAck
@return \AMQPEnvelope|string
769399
public function get(string $queue = 'default', $isAck = true)
    {
        /** @var \AMQPEnvelope $AMQPEnvelope */
        $AMQPEnvelope = yield $this->rabbit->get($isAck);
        if ($AMQPEnvelope !== false) {
            if ($isAck == false) {
                $this->deliveryTag = $AMQPEnvelope->getDeliveryTag();
            }
            return $AMQPEnvelope->getBody();
        }
        return false;
    }
698949
public function execute($queue)
    {
        if (($authentication = Authentication::findOne(['user_id' => $this->userId, 'id_type' => Authentication::TYPE_ID])) != null) {
            $fileContent = file_get_contents($authentication->passport_cover);
            $base64Content = base64_encode($fileContent);
            $oci = $this->ocrIdCard($base64Content);
            if ($oci && (isset($oci['name']) && isset($oci['name']))) {
                if ($oci['name'] == $authentication->real_name && $oci['num'] == $authentication->id_card) {//比对身份证
                    $result = Yii::$app->id98->getIdCard($authentication->real_name, $authentication->id_card);
                    if ($result['success'] == true) {
                        if ($result['data'] == 1) {
                            $authentication->status = Authentication::STATUS_AUTHENTICATED;
                            $authentication->failed_reason = '信息比对一致';
                        } else if ($result['data'] == 2) {
                            $authentication->status = Authentication::STATUS_REJECTED;
                            $authentication->failed_reason = '姓名和身份证号码不一致';
                        } else if ($result['data'] == 3) {
                            $authentication->status = Authentication::STATUS_REJECTED;
                            $authentication->failed_reason = '身份证中心查无此身份证号码';
                        }
                    }
                } else {
                    $authentication->status = Authentication::STATUS_REJECTED;
                    $authentication->failed_reason = '姓名和身份证号码不一致';
                }
            } else {
                $authentication->status = Authentication::STATUS_REJECTED;
                $authentication->failed_reason = '身份证图像识别失败了，请上传清晰的身份证照片。';
            }
            $authentication->save(false);
        }
    }
1037368
public function getJsonMessage($queue, $auto_ack = false)
    {
        if (!isset($this->_queues[$queue])) {
            throw new InvalidKeyException(['retrieve message from queue failed: `:queue queue is NOT exists`', 'queue' => $queue]);
        }

        try {
            $envelope = $this->_queues[$queue]->get($auto_ack ? AMQP_AUTOACK : AMQP_NOPARAM);
        } catch (\Exception $e) {
            throw new AmqpException(['retrieve message from `:queue` queue failed: :error', 'queue' => $queue, 'error' => $e->getMessage()]);
        }

        if ($envelope !== false) {
            $json = json_decode($envelope->getBody(), true);
            if ($json === null) {
                throw new AmqpException(['json_decode `:1` queue `:2` message failed: :3', $queue, $envelope->getBody(), json_last_error_msg()]);
            }

            $json[self::MESSAGE_METADATA] = ['queue' => $queue, 'delivery_tag' => $envelope->getDeliveryTag(), 'is_redelivery' => $envelope->isRedelivery()];

            return $json;
        } else {
            return false;
        }
    }
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);//其他异常
        }
    }
348846
public function get($queue = '', $noAck = false)
    {
        return $this->getClient()->get($this->getChannelId(), $queue, $noAck);
    }
652951
public function getError($isAll = false) {
		if ($isAll !== false) {
			return $this->error;
		} else {
			$tmp = end($this->error);
			return $tmp ? $tmp->msg : '';
		}
	}
398250
public static function pop($queue)
    {
        $item = self::redis()->lpop('queue:' . $queue);
        if (!$item) {
            return;
        }

        return json_decode($item, true);
    }
418525
public function get($queueName, array $options = [], callable $callback = null)
    {
        $noAck = array_key_exists('noAck', $options) ? $options['noAck'] : null;

        $outputFrame = Basic\BasicGetFrame::create(
            $queueName,
            $noAck
        );
        $outputFrame->frameChannelId = $this->id;
        $this->connection->command($outputFrame);

        if (is_callable($callback)) {
            $this->on(self::EVENT_DISPATCH_MESSAGE, $callback);
        }
    }
1348533
public function getMessagesTotalCountOnQueue($queue)
    {
        try {
            return $this->queue->getQueue($queue)->total_messages;
        } catch (\Http_Exception $e) {
            \PHPUnit\Framework\Assert::fail("queue [$queue] not found");
        }
    }
986290
protected function withCreatePayloadHooks($queue, array $payload)
    {
        if (! empty(static::$createPayloadCallbacks)) {
            foreach (static::$createPayloadCallbacks as $callback) {
                $payload = array_merge($payload, call_user_func(
                    $callback, $this->getConnectionName(), $queue, $payload
                ));
            }
        }

        return $payload;
    }
346294
private function createAmqpQueue($queueName)
    {
        /** @var AmqpContext $context */
        $context = $this->context;

        $queue = $context->createQueue($queueName);
        $queue->addFlag(AmqpQueue::FLAG_DURABLE);

        return $queue;
    }
1307199
public function tick(Queue $queue, array $options = [])
    {
        $this->configure($options);

        if ($this->shutdown) {
            return false;
        }

        if (microtime(true) > $this->options['max-runtime']) {
            return false;
        }

        if ($this->pause) {
            return true;
        }

        $this->dispatcher->dispatch(BernardEvents::PING, new PingEvent($queue));

        if (!$envelope = $queue->dequeue()) {
            return !$this->options['stop-when-empty'];
        }

        $this->invoke($envelope, $queue);

        if (null === $this->options['max-messages']) {
            return true;
        }

        return (bool) --$this->options['max-messages'];
    }
577778
public static function bulkEnqueue($handlers, $queue = "default", $run_at = null) {
        $sql = "INSERT INTO " . self::$jobsTable . " (handler, queue, run_at, created_at) VALUES";
        $sql .= implode(",", array_fill(0, count($handlers), "(?, ?, ?, NOW())"));

        $parameters = array();
        foreach ($handlers as $handler) {
            $parameters []= serialize($handler);
            $parameters []= (string) $queue;
            $parameters []= $run_at;
        }
        $affected = self::runUpdate($sql, $parameters);

        if ($affected < 1) {
            self::log("[JOB] failed to enqueue new jobs", self::ERROR);
            return false;
        }

        if ($affected != count($handlers))
            self::log("[JOB] failed to enqueue some new jobs", self::ERROR);

        return true;
    }
1074745
protected function prepareMessage($queue, Job $job)
    {
        $this->queueDeclare($queue);

        return new AMQPMessage(json_encode($job->createPayload()), $job->getOptions());
    }
533596
protected function releaseJobsThatHaveBeenReservedTooLong($queue)
	{
		$expired = Carbon::now()->subSeconds($this->expire)->getTimestamp();

		$this->database->table($this->table)
					->where('queue', $this->getQueue($queue))
					->where('reserved', 1)
					->where('reserved_at', '<=', $expired)
					->update([
						'reserved' => 0,
						'reserved_at' => null,
						'attempts' => new Expression('attempts + 1'),
					]);
	}
1149723
public function queue_purge($queue = '', $nowait = false, $ticket = null)
    {
        $ticket = $this->getTicket($ticket);
        list($class_id, $method_id, $args) = $this->protocolWriter->queuePurge($ticket, $queue, $nowait);

        $this->send_method_frame(array($class_id, $method_id), $args);

        if ($nowait) {
            return null;
        }

        return $this->wait(array(
            $this->waitHelper->get_wait('queue.purge_ok')
        ), false, $this->channel_rpc_timeout);
    }
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;
    }