如何让 AsyncQueue 支持 Closure

[huangdijia]

用过 Laravel 的小伙伴都知道，Laravel 有个助手函数 dispatch，用于快速投递 Job，用法很简单：

dispatch(new ExampleJob());

在某个版本增加了对 Closure 支持，这意味着不需要定义 Job，随时随地可以对任意一段业务逻辑以 Job 的方式投递到队列，用法如下：

$podcast = App\Podcast::find(1);

dispatch(function () use ($podcast) {
    $podcast->publish();
});

是不是很帅？是不是在 Hyperf 里也想这样玩？

目前 Hyperf 助手函数还没有这个函数，我们定义一个

    /**
     * @param \Hyperf\AsyncQueue\JobInterface $job
     * @throws TypeError
     * @throws InvalidDriverException
     * @throws InvalidArgumentException
     * @return bool
     * @version 1.0
     */
    function dispatch($job, ...$arguments)
    {
        if ($job instanceof \Hyperf\AsyncQueue\JobInterface) {
            $container = \Hyperf\Utils\ApplicationContext::getContainer();
            /** @var \Hyperf\AsyncQueue\Driver\DriverInterface $driver */
            $driver = $container->get(\Hyperf\AsyncQueue\Driver\DriverFactory::class)->get((string) ($arguments[0] ?? $job->queue ?? 'default'));
            return $driver->push($job, (int) ($arguments[1] ?? $job->delay ?? 0));
        }

        throw new \InvalidArgumentException('Not Support job type.');
    }

这只是实现了第一步，可以快速投递 Job 了，还需要对 Closure 实现支持。

参考了 AnnotationJob 实现一个支持 Closure 的 Job：

<?php
namespace App\AsyncQueue;

use Closure;
use Hyperf\AsyncQueue\Job;
use Hyperf\Contract\NormalizerInterface;
use Hyperf\Di\ClosureDefinitionCollectorInterface;
use Hyperf\Utils\ApplicationContext;
use InvalidArgumentException as GlobalInvalidArgumentException;
use Opis\Closure\SerializableClosure;
use Psr\Container\ContainerInterface;
use ReflectionFunction;

class ClosureJob extends Job
{
    /**
     * @var string
     */
    public $class;

    /**
     * @var string
     */
    public $method;

    /**
     * @var SerializableClosure
     */
    protected $closure;

    public function __construct(Closure $closure, int $maxAttempts = 0)
    {
        $this->closure = new SerializableClosure($closure);
        $this->maxAttempts = $maxAttempts;
        $this->class = 'Closure';
        $this->method = with(new ReflectionFunction($this->closure->getClosure()), function ($reflection) {
            return sprintf('%s:%s', str_replace(rtrim(BASE_PATH, '/') . '/', '', $reflection->getFileName()), $reflection->getStartLine());
        });
    }

    public function handle()
    {
        $closure = $this->closure->getClosure();
        $parameters = $this->parseClosureParameters($closure, []);

        call($closure, $parameters);
    }

    protected function getContainer(): ContainerInterface
    {
        return ApplicationContext::getContainer();
    }

    protected function getClosureDefinitionCollector(): ClosureDefinitionCollectorInterface
    {
        return $this->getContainer()->get(ClosureDefinitionCollectorInterface::class);
    }

    protected function getNormalizer(): NormalizerInterface
    {
        return $this->getContainer()->get(NormalizerInterface::class);
    }

    /**
     * @throws GlobalInvalidArgumentException
     */
    protected function parseClosureParameters(Closure $closure, array $arguments): array
    {
        if (! $this->getContainer()->has(ClosureDefinitionCollectorInterface::class)) {
            return [];
        }

        $definitions = $this->getClosureDefinitionCollector()->getParameters($closure);

        return $this->getInjections($definitions, 'Closure', $arguments);
    }

    /**
     * @throws GlobalInvalidArgumentException
     */
    protected function getInjections(array $definitions, string $callableName, array $arguments): array
    {
        $injections = [];

        foreach ($definitions ?? [] as $pos => $definition) {
            $value = $arguments[$pos] ?? $arguments[$definition->getMeta('name')] ?? null;
            if ($value === null) {
                if ($definition->getMeta('defaultValueAvailable')) {
                    $injections[] = $definition->getMeta('defaultValue');
                } elseif ($definition->allowsNull()) {
                    $injections[] = null;
                } elseif ($this->getContainer()->has($definition->getName())) {
                    $injections[] = $this->getContainer()->get($definition->getName());
                } else {
                    throw new \InvalidArgumentException("Parameter '{$definition->getMeta('name')}' "
                        . "of {$callableName} should not be null");
                }
            } else {
                $injections[] = $this->getNormalizer()->denormalize($value, $definition->getName());
            }
        }

        return $injections;
    }
}

最后，对 dispatch() 再稍微改造一下：

    /**
     * @param Closure|\Hyperf\AsyncQueue\JobInterface $job
     * @throws TypeError
     * @throws InvalidDriverException
     * @throws InvalidArgumentException
     * @return bool
     * @version 2.0
     */
    function dispatch($job, ...$arguments)
    {
        if ($job instanceof Closure) {
            $job = new \App\AsyncQueue\ClosureJob($job, (int) ($arguments[3] ?? 0));
        }

        if ($job instanceof \Hyperf\AsyncQueue\JobInterface) {
            $container = \Hyperf\Utils\ApplicationContext::getContainer();
            /** @var \Hyperf\AsyncQueue\Driver\DriverInterface $driver */
            $driver = $container->get(\Hyperf\AsyncQueue\Driver\DriverFactory::class)->get((string) ($arguments[0] ?? $job->queue ?? 'default'));
            return $driver->push($job, (int) ($arguments[2] ?? $job->delay ?? 0));
        }

        throw new \InvalidArgumentException('Not Support job type.');
    }

最最最后，让我们测试一下：

dispatch(function() {
    // 业务逻辑
});