在指定的时间后执行函数
一次性执行
@param int $msec
@param callable $callback
@return int
183920
public function after(int $msec, callable $callback)
    {
        // 清除旧定时器
        $this->clear();
        // 外部获取协程id
        $tid = Coroutine::tid();
        $top = $tid == Coroutine::id();
        // 设置定时器
        $timerId = swoole_timer_after($msec, function () use ($callback, $tid, $top) {
            // 创建协程
            Coroutine::go($callback, [], $tid, $top);
        });
        // 保存id
        $this->_timerId = $timerId;
        // 返回
        return $timerId;
    }
215412
public function max($callback = null)
    {
        return $this->reduce(function ($carry, $value, $key) use ($callback) {
            if ($callback === null) {
                $result = $value;
            } else {
                $result = $this->accessKey($value, $callback, $key);
            }

            if ($carry === null || $carry < $result) {
                return $result;
            }

            return $carry;
        }, null);
    }
342125
public function execute($command, callable $callback = null)
    {
        $this->login();

        return $this->ssh->exec(sprintf('cd %s; %s', $this->directory, $command), $callback);
    }
1205009
protected function runProcess($callback = null)
    {
        $process = $this->processBuilder
            ->setArguments(
                $this->prepareArguments()
            )
            ->getProcess();

        // exit code
        $result = $process->run($callback);

        // text-mode
        if (null === $callback) {
            $result = $process->getOutput();

            if (false === $process->isSuccessful()) {
                $process->getErrorOutput();
            }
        }

        return $result;
    }
230897
public function run($callback)
    {
        $attempt = 0;
        $try = true;

        while ($try) {

            $result = null;
            $exception = null;

            $this->wait($attempt);
            try {
                $result = call_user_func($callback);
            } catch (Exception $e) {
                $this->exceptions[] = $e;
                $exception = $e;
            }
            $try = call_user_func($this->decider, ++$attempt, $this->getMaxAttempts(), $result, $exception);

            if($try && isset($this->errorHandler)) {
                call_user_func($this->errorHandler, $exception, $attempt, $this->getMaxAttempts());
            }
        }

        return $result;
    }
1331846
public function each( callable $callback, $count = 1000 )
	{
		if ( is_null( $this->orders ) && is_null( $this->unionOrders ) )
			throw new RuntimeException( 'You must specify an orderBy clause when using the "each" function.' );

		return $this->chunk( $count, function ( $results ) use ( $callback )
		{
			foreach ( $results as $key => $value )
				if ( $callback( $value, $key ) === false )
					return false;
		} );
	}
213098
public function countBy($callback = null)
    {
        if (is_null($callback)) {
            $callback = function ($value) {
                return $value;
            };
        }

        return new static($this->groupBy($callback)->map(function ($value) {
            return $value->count();
        }));
    }
803756
public function inputObserver($callback, $exitCicle = null)
    {
        if (self::isCli() === false || is_callable($callback) === false) {
            return false;
        }

        $this->io = $callback;
        $this->ec = $exitCicle ? $exitCicle : $this->ec;

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

        $this->started = true;
        $this->fireInputObserver();

        return true;
    }
391174
public function eachById(callable $callback, $count = 1000, $column = null, $alias = null)
    {
        return $this->chunkById($count, function ($results) use ($callback) {
            foreach ($results as $key => $value) {
                if ($callback($value, $key) === false) {
                    return false;
                }
            }
        }, $column, $alias);
    }
1185074
public function bindForCallback(string $abstract, $concrete, callable $callback)
    {
        $hasOriginal      = $this->container->bound($abstract);
        $originalInstance = false;

        if ($hasOriginal) {
            $binding = $this->container->getBindings()[$abstract] ?? null;

            $hasInstance = $binding && $binding['shared'] && $this->container->resolved($abstract);

            if ($hasInstance) {
                $originalInstance = $this->container->make($abstract);
            }
        }

        $this->container->instance($abstract, $concrete);

        try {
            $return = $callback();
            return $return;
        } finally {
            unset($this->container[$abstract]);

            if ($hasOriginal && (isset($binding) && $binding)) {
                $this->container->bind($abstract, $binding['concrete'], $binding['shared']);

                if ($originalInstance) {
                    $this->container->instance($abstract, $originalInstance);
                }
            }
        }
    }
39382
public function sum($callback)
    {
        if (is_string($callback)) {
            $callback = $this->valueRetriever($callback);
        }

        return $this->reduce(function($result, $item) use ($callback) {
            return $result += $callback($item);
        }, 0);
    }
274563
public function exec($queue_name,$max,$phpExec=''){
        $max = $max ? $max : 1;

        $time = time();
        while(1) {
            //执行死循环
            try {
                while ($this->threadRunning < $max && !Queue::instance()->is_end($queue_name)) {
                    $this->running[] = new Thread($queue_name,$phpExec);
                    usleep(200000);
                    $this->threadRunning++;
                }
            }
            catch (Exception $e) {
                switch($e->getCode()) {
                case 30001:
                    logger::emergency(sprintf('ERROR:%d @ %s', $e->getCode(), $e->getMessage));
                    exit;
                }
            }

            //检查是否已经结束
            if ($this->threadRunning == 0) {

                break;
            }

            //等待代码执行完成
            usleep(50000);

            $thread_close = array();//记录线程的关闭状态

            //检查已经完成的任务
            foreach ($this->running as $idx => $thread) {

                if (!$thread->isRunning() || $thread->isOverExecuted($max)) {
                    $thread_close[] = proc_close($thread->resource);//记录线程的关闭状态
                    unset($this->running[$idx]);
                    $this->threadRunning--;
                }
            }
        }
    }
797617
public function login($callback)
    {
        // 兼容二维码位置传参
        if (is_string($callback)) {
            /* @deprecated 后面会废除此用法，建议传闭包自己处理二维码 */
            $callback = function ($qrcode) use ($callback) {
                Utils::getFilesystem()->dumpFile($callback, $qrcode);
            };
        }

        $resolver = new CredentialResolver($this);
        // 进行授权流程，确认授权
        $resolver->resolve($callback);
        // 等待授权验证成功
        $credential = $resolver->wait();
        $this->setCredential($credential);
        //获取在线状态避免103
        $this->getFriendsOnlineStatus();

        return $credential;
    }
1263130
public function find(callable $callback, $keepKey = false, $offset = null, $limit = null)
    {
        return $this->bindNewInstance(Arr::find($this->convertArray($this), $callback, $keepKey, $offset, $limit));
    }
339054
public function command($command, $method) {

        self::$commands[] = $command;
        self::$callbacks[$command] = $method;
    }
761114
public function executeCommand(CommandInterface $command, callable $callback)
    {
        $this->connection->executeCommand($command, $this->wrapCallback($callback));
    }
743106
public function tick($ms, $callback, $user_param = null)
    {
        $id = \swoole_timer_tick($ms, function ($user_param_one) use ($callback) {
            try {
                $callback($user_param_one);
            } catch (\Throwable $e) {
                displayExceptionHandler($e);
            }
        }, $user_param);
        $this->timerIdArr[$id] = $id;
        return $id;
    }