执行具体的任务

@param string $queue_name 队列名称
@param int $max 最大可开启的进程数
@param String $phpExec PHP脚本路径
760043
public function exec($queue_name,$max=5,$phpExec=''){
        $this->__consumer->exec($queue_name,$max,$phpExec);
    }
828226
public function actionMessages($queue_name, $subscriber_id = null)
    {
        if (($subscriber_id = trim($subscriber_id)) === '') {
            $subscriber_id = null;
        }
        /** @var QueueInterface $queue */
        list($queue, $authItems) = $this->loadQueue($queue_name, ['nfy.message.read', 'nfy.message.create']);
        $this->verifySubscriber($queue, $subscriber_id);

        $formModel = new models\MessageForm();
        if ($authItems['nfy.message.create'] && isset($_POST['MessageForm'])) {
            $formModel->attributes = $_POST['MessageForm'];
            if ($formModel->validate()) {
                $queue->send($formModel->content, $formModel->category);

                return $this->redirect([
                    'messages',
                    'queue_name' => $queue_name,
                    'subscriber_id' => $subscriber_id,
                ]);
            }
        }

        $dataProvider = null;
        if ($authItems['nfy.message.read']) {
            $dataProvider = new \yii\data\ArrayDataProvider([
                'allModels' => $queue->peek($subscriber_id, 200, [
                    components\Message::AVAILABLE,
                    components\Message::RESERVED,
                    components\Message::DELETED,
                ]),
                'sort' => [
                    'attributes' => ['id'],
                    'defaultOrder' => ['id' => SORT_DESC],
                ],
            ]);
            // reverse display order to simulate a chat window, where latest message is right above the message form
            $dataProvider->setModels(array_reverse($dataProvider->getModels()));
        }

        return $this->render('messages', [
            'queue' => $queue,
            'queue_name' => $queue_name,
            'dataProvider' => $dataProvider,
            'model' => $formModel,
            'authItems' => $authItems,
        ]);
    }
1070681
public function tick(int $msec, callable $callback)
    {
        // 清除旧定时器
        $this->clear();
        // 外部获取协程id
        $tid = Coroutine::tid();
        $top = $tid == Coroutine::id();
        // 设置定时器
        $timerId = swoole_timer_tick($msec, function () use ($callback, $tid, $top) {
            // 创建协程
            Coroutine::go($callback, [], $tid, $top);
        });
        // 保存id
        $this->_timerId = $timerId;
        // 返回
        return $timerId;
    }
1091254
public function delete()
    {
        if (!$this->queueName) {
            throw new \Exception('QueueName 队列名称不存在');
        }

        if (!$this->message) {
            throw new \Exception('Message 消息不存在');
        }

        $params = [
            'queueName'     => $this->queueName,
            'receiptHandle' => $this->message->get('receiptHandle'),
        ];

        $request = new DeleteMessageRequest($params);

        try {
            $this->client->send($request);
        } catch (RequestException $e) {
            throw $e;
        } catch (ResponseException $e) {
            $message = $e->getBody();

            if (!$message->has('code')) {
                throw $e;
            } else if (!in_array($message->get('code'), ['4430', '4440'])) {    // 4430：句柄无效，4440：队列不存在
                throw $e;
            }
        } catch (\Exception $e) {
            throw $e;
        }
    }
25384
protected function initialization($controller_name, $method_name)
    {
        if ($this->isRPC && !empty($this->rpc_request_id)) {
            //全链路监控保证调用的request_id唯一
            $this->context['request_id'] = $this->rpc_request_id;
        } else {
            $this->context['request_id'] = time() . crc32($controller_name . $method_name . getTickTime() . rand(1, 10000000));
        }
        $this->context['controller_name'] = $controller_name;
        $this->context['method_name'] = "$controller_name::$method_name";
        $this->context['ip'] = $this->getFdInfo()['remote_ip'];
        if (!empty($this->uid)) {
            $this->context['uid'] = $this->uid;
        }
        $this->db = $this->loader->mysql("mysqlPool",$this);
        $this->redis = $this->loader->redis("redisPool",$this);
        if ($this->isEnableError) {
            $this->Error = $this->loader->model(Error::class, $this);
        }
    }
219439
function SetMaxQueryTime ( $max )
    {
        assert ( is_int($max) );
        assert ( $max>=0 );
        $this->_maxquerytime = $max;
    }
287986
public static function buildPHPExtension($name, $min = null, $max = null, $optional = false)
    {
        $returnValue = null;

        
        $returnValue = new common_configuration_PHPExtension($min, $max, $name, $optional);
        

        return $returnValue;
    }
1084538
public void setMaxCapacity(String queue,float maxCapacity) {
      rmConf.setFloat(
        toFullPropertyName(queue, MAX_CAPACITY_PROPERTY), maxCapacity);
  }
142549
public function createQueue(string $queueName, array $attributes=array()) {
        if(!$queueName){
            return $this->error('参数有误');
        }
        $request = new CreateQueueRequest($queueName, $this->makeQueueAttributes($attributes));
        try{
            $this->client->createQueue($request);
            return $this->success('创建队列成功');
        }catch(MnsException $exception){
            return $this->parseException($exception, '创建队列');
        }
    }
494860
public function deleteQueue($name, array $options = [], callable $callback = null)
    {
        $ifUnused = array_key_exists('ifUnused', $options) ? (bool)$options['ifUnused'] : null;
        $ifEmpty = array_key_exists('ifEmpty', $options) ? (bool)$options['ifEmpty'] : null;
        $noWait = array_key_exists('noWait', $options) ? (bool)$options['noWait'] : null;

        $outputFrame = Queue\QueueDeleteFrame::create($name, $ifUnused, $ifEmpty, $noWait);
        $outputFrame->frameChannelId = $this->id;
        $this->connection->command($outputFrame);

        if (is_callable($callback)) {
            $this->on(self::EVENT_ON_CHANNEL_DELETE_QUEUE_CALLBACK, $callback);
        }
    }
785251
public function daemon( $connectionName, $queue = null, $delay = 0, $memory = 128, $sleep = 3, $maxTries = 0 )
	{
		$lastRestart = $this->getTimestampOfLastQueueRestart();

		while ( true )
		{
			if ( $this->daemonShouldRun() )
			{
				$this->runNextJobForDaemon( $connectionName, $queue, $delay, $sleep, $maxTries );
			}
			else
			{
				$this->sleep( $sleep );
			}

			if ( $this->memoryExceeded( $memory ) || $this->queueShouldRestart( $lastRestart ) )
			{
				$this->stop();
			}
		}
	}
950641
public function preparePool($classOrName) {
        $queue = new \SplQueue();

        $this->queues[ $classOrName ] = $queue;
        /* @var $bean PoolAble|PoolTrait */
        $bean = Ioc::beanCreate($classOrName, false);
        $config = $bean->poolConfig();
        $bean->_poolName_ = $classOrName;
        $min = $config[ 'min' ];
        $max = $config[ 'max' ];

        $queue->push($bean);
        for ($i = 1; $i < $min; $i++) {
            $bean = Ioc::beanCreate($classOrName, false);
            $bean->_poolName_ = $classOrName;
            $queue->push($bean);
        }
        if (!$max || $max <= $min) {
            $max = $min + 1;
        }

        $chanel = new Channel($max - $min + 1);
        $this->channels[ $classOrName ] = $chanel;
        $buffers = [];
        for ($i = 0; $i < $max - $min; $i++) {
            $buffer = new PoolBuffer($classOrName);
            $chanel->push($buffer);
            $buffers[] = $buffer;
        }
        $this->buffers[ $classOrName ] = $buffers;
        //定时删除
        $idle = $config[ 'idle' ];
        if (!$idle) {
            $idle = 60;
        }
        $check = $config[ 'check' ];
        if (!$check) {
            $check = 30;
        }

        swoole_timer_tick(1000 * $check, function()use($idle) {
            foreach ($this->buffers as $class => $buffer_array) {
                foreach ($buffer_array as $buffer) {
                    /* @var $buffer PoolBuffer */
                    if ($buffer->is_use) {
                        continue;
                    }
                    $time = time() - $buffer->lastActiveTime;
                    if ($time > $idle) {
                        unset($buffer->bean);
                        $buffer->bean = null;
                    }
                }
            }
        });
    }
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);
    }
840326
public function read ($list_name, $limit = NULL, $query = NULL, $view = NULL, $sort = NULL, $options = NULL) {
		// Check limit is set
		if ($limit < 1 || is_null($limit)) {
			$limit = $this->MAX_ROWS;
		}

		// Create Query XML is query is being used
		$xml_options = '';
		$xml_query   = '';
		$fields_xml = '';

		// Setup Options
		if ($query instanceof Service\QueryObjectService) {
			$xml_query = $query->getCAML();
			$xml_options = $query->getOptionCAML();
		} else {

			if (!is_null($query)) {
				$xml_query .= $this->whereXML($query); // Build Query
			}
			if (!is_null($sort)) {
				$xml_query .= $this->sortXML($sort);// add sort
			}

			// Add view or fields
			if (!is_null($view)){
				// array, fields have been specified
				if(is_array($view)){
					$xml_options .= $this->viewFieldsXML($view);
				}else{
					$xml_options .= '<viewName>' . $view . '</viewName>';
				}
			}
		}

		// If query is required
		if (!empty($xml_query)) {
			$xml_options .= '<query><Query>' . $xml_query . '</Query></query>';
		}

		/*
		 * Setup basic XML for querying a SharePoint list.
		 * If rowLimit is not provided SharePoint will default to a limit of 100 items.
		 */
		$CAML = '
			<GetListItems xmlns="http://schemas.microsoft.com/sharepoint/soap/">
				<listName>' . $list_name . '</listName>
				<rowLimit>' . $limit . '</rowLimit>
				' . $xml_options . '
				<queryOptions xmlns:SOAPSDK9="http://schemas.microsoft.com/sharepoint/soap/" >
					<QueryOptions>
						' . $options . '
					</QueryOptions>
				</queryOptions>
			</GetListItems>';

		// Ready XML
		$xmlvar = new \SoapVar($CAML, XSD_ANYXML);
		$result = NULL;

		// Attempt to query SharePoint
		try {
			$result = $this->xmlHandler($this->soapClient->GetListItems($xmlvar)->GetListItemsResult->any);
		} catch (\SoapFault $fault) {
			$this->onError($fault);
		}

		// Return a XML as nice clean Array
		return $result;
	}
128763
public function createQueue($queueName)
    {
        if ($this->queueExists($queueName)) {
            return;
        }

        $parameters = [
            'QueueName' => $queueName,
        ];

        if ($this->isFifoQueue($queueName)) {
            $parameters['Attributes'] = [
                'FifoQueue' => 'true',
            ];
        }

        $result = $this->sqs->createQueue($parameters);

        $this->queueUrls[$queueName] = $result['QueueUrl'];
    }
629215
private function create($name, $timeout = null)
    {
        if(null === $timeout) {
            $timeout = 10000;
        }

        $sql = 'INSERT INTO ' . $this->queueTable . '
            (queue_name, timeout)
            VALUES
            (:queue_name, :timeout)
            ';
        $stmt = $this->getDb()->prepare($sql);
        $stmt->bindParam(':queue_name', $name, \PDO::PARAM_STR);
        $stmt->bindParam(':timeout', $timeout, \PDO::PARAM_INT);
        $stmt->execute();
        return true;
    }
507196
function ruleMax($max, $error_message = null) {
        if (!is_numeric($max)) {
            throw new \Exception('Incorrect parameter MAX in rule max');
        }

        return $this->_rule('max:' . $max, $error_message);
    }