php - 如何在 Symfony 3 中实现异步队列运行方法

Question

首先，关于我的项目的一些基本信息:我有一个使用 Symfony 3 构建的网站。对于某些任务，我正在考虑实现以运行异步 PHP 方法。有些事件会占用大量时间，但其结果不需要立即显现。

例如:在方法 newOrder 中，我有函数 addUserLTV，它只执行几个步骤。客户不必等待所有步骤完成，只需在基本操作后立即获得确认 - 'newOrder' 将添加 addUserLTV 到队列并立即显示确认(完成运行)。队列任务将在服务器有时间执行时运行。

public function addUserLTV( $userID, $addLTV )
{ //same code
}

怎么做？在 symphony 3 中有可能吗？

Answer 1

这是您可以使用 enqueue bundle 轻松完成的事情.关于为什么要选择它的几句话:

• 支持a lot of transports从最简单的系统(文件系统)到企业级系统(RabbitMQ 或 Amazon SQS)。
• 它带有一个非常强大的 bundle .
• 它具有顶级抽象，可以最轻松地使用。
• 还有很多可能会派上用场。

关于你的问题。以下是使用入队包执行此操作的方法。关注setup instructions from the doc .

现在 addUserLTV 方法将如下所示:

<?php
namespace Acme;

use Enqueue\Client\ProducerInterface;

class AddUserLTVService
{
    /**
     * @var ProducerInterface
     */
    private $producer;

    /**
     * @param ProducerInterface $producer
     */
    public function __construct(ProducerInterface $producer)
    {
        $this->producer = $producer;
    }

    public function addUserLTV( $userID, $addLTV )
    {
        $this->producer->sendCommand('add_user_ltv', [
            'userId' => $userID, 
            'ltv' => $addLTV]
        );
    }
}

它使用客户端(我之前提到过的顶级抽象)将消息发送到消息队列。该服务必须注册到 Symfony 容器:

services:
    Acme\AddUserLTVService: 
        arguments: ['@enqueue.producer']

现在让我们看看消费端。您需要一个命令处理器来完成这项工作:

<?php
namespace Acme;

use Enqueue\Client\CommandSubscriberInterface;
use Enqueue\Psr\PsrContext;
use Enqueue\Psr\PsrMessage;
use Enqueue\Psr\PsrProcessor;
use Enqueue\Util\JSON;

class AddUserTVAProcessor implements PsrProcessor, CommandSubscriberInterface
{
    public function process(PsrMessage $message, PsrContext $context)
    {
        $data = JSON::decode($message->getBody());

        $userID = $data['userID'];
        $addLTV = $data['ltv'];

        // do job

        return self::ACK;
    }

    public static function getSubscribedCommand()
    {
        return 'add_user_ltv';
    }
}

使用 enqueue.client.processor 标签将其注册为服务:

services:
    Acme\AddUserTVAProcessor: 
        tags:
            - {name: 'enqueue.client.processor'}

这就是编码。运行 consume 命令，你就完成了:

./bin/console enqueue:consume --setup-broker -vvv