下载头像并保存
@param \yii\queue\Queue $queue
@throws \yii\base\ErrorException
@throws \yii\base\Exception
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);
            };
        }
    }
37389
protected function queueDeclare($queue)
    {
        $this->channel->queue_declare($queue, false, true, false, false);
    }
83008
public function seeQueueHasTotalCount($queue, $expected)
    {
        $this->assertEquals($expected, $this->queueDriver->getMessagesTotalCountOnQueue($queue));
    }
780419
public function deleteQueue(string $queueName) {
        if(!$queueName){
            return $this->error('参数有误');
        }
        try{
            $this->client->deleteQueue($queueName);
            return $this->success('删除队列成功');
        }catch(MnsException $exception){
            return $this->parseException($exception, '删除队列');
        }
    }
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);
    }
247106
public function entityQueueCreate($queue, $destBid) {
    $entity_subqueue = $this->entityManager->getStorage('entity_subqueue')->load($queue);
    $items = $entity_subqueue->get('items')->getValue();
    $items[] = ['target_id' => $destBid[0]];
    $entity_subqueue->set('items', $items);
    $entity_subqueue->save();
  }
35859
public function deleteReserved($queue, $job)
    {
        $this->getConnection()->zrem($this->getQueue($queue).':reserved', $job);
    }
366962
public function pop($queue = null)
    {
        return tap(parent::pop($queue), function ($result) use ($queue) {
            if ($result) {
                $this->event($this->getQueue($queue), new JobReserved($result->getReservedJob()));
            }
        });
    }
496240
public function setEventQueue(Event $event, Queue $queue)
    {
        $mask = $this->getConnectionMask($event->getConnection());
        $this->getEventQueueDeferred($mask)->resolve($queue);
    }
562605
public function setQueue( $queue )
	{
		$this->queue = $queue;

		if ( $this->mailer )
		{
			$this->mailer->setQueue( $queue );
		}

		return $this;
	}
546746
public function deleteReserved($queue, $job)
	{
		$this->redis->zrem($this->getQueue($queue).':reserved', $job);
	}
343887
public function reset($queue)
    {
        if (!function_exists('posix_getsid')) {
            throw JobConfigurationException::create('Function posix_getsid don\'t exists');
        }

        $configurations = $this->configurationManager->findPotentialDeadJobs($queue);

        foreach ($configurations as $configuration) {
            if ($this->restoreManager->reset($configuration)) {
                return true;
            }
        }

        return false;
    }
1008628
public function setQueue(QueueContract $queue)
    {
        if ($this->mailer instanceof MailerContract) {
            $this->mailer->setQueue($queue);
        }

        $this->queue = $queue;

        return $this;
    }
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);//其他异常
        }
    }
291112
public function setQueueOptions(array $options = [], $queue = self::PRIORITY_NORM)
    {
        msg_set_queue($this->queues[$queue], $options);
    }
279150
private function runQueue($queue, $worker)
    {
        $result['count'] = 0;
        $result['total'] = $queue->numberOfItems();
        while ($item = $queue->claimItem()) {
            try {
                $worker->processItem($item->data);
                $queue->deleteItem($item);
                $result['count']++;
            } catch (SuspendQueueException $e) {
                $queue->releaseItem($item);
                $result['error'] = $e;
            }
        }

        return $result;
    }
443578
public function pop($queue = null)
	{
		$queue = $this->getQueue($queue);

		if ( ! is_null($this->expire))
		{
			$this->releaseJobsThatHaveBeenReservedTooLong($queue);
		}

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

			$this->database->commit();

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

		$this->database->commit();
	}