Hyperf-消息隊列

Hyperf-消息隊列

     官方文檔里面有詳細說明，安裝和配置不再贅述，這里只是記錄下實際中Redis異步隊列的具體使用。

     注意：異步隊列區別於 RabbitMQ Kafka 等消息隊列，它只提供一種 異步處理 和 異步延時處理 的能力，並不能嚴格地保證消息的持久化和不支持完備的ACK 應答機制。

     工作原理

     ConsumerProcess 是異步消費進程，會根據用戶創建的 Job 或者使用 @AsyncQueueMessage 的代碼塊，執行消費邏輯。 Job 和 @AsyncQueueMessage 都是需要投遞和執行的任務，即數據、消費邏輯都會在任務中定義。

     1）Job 類中成員變量即為待消費的數據，handle() 方法則為消費邏輯。
     2）@AsyncQueueMessage 注解的方法，構造函數傳入的數據即為待消費的數據，方法體則為消費邏輯。

 

    1、配置異步消費進程

    config/autoload/processes.php

<?php

  return [

       Hyperf\AsyncQueue\Process\ConsumerProcess::class,
  ];

    2、定義Job

    Job 類中成員變量即為待消費的數據，handle() 方法則為消費邏輯

    app/Job/AnalyseJob.php

<?php

declare(strict_types=1);

namespace App\Job;

use App\Log;
use App\Service\V1\Order\OrderService;
use Hyperf\AsyncQueue\Job;

class AnalyseJob extends Job
{
    public $params;

    /**
     * 任務執行失敗后的重試次數，即最大執行次數為 $maxAttempts+1 次
     * @var int
     */
    protected $maxAttempts = 2;

    public function __construct($params)
    {
        $this->params = $params;
    }

    public function handle()
    {
        // 這里的邏輯會在 ConsumerProcess 進程中執行
        try {
            var_dump(">>> 監聽到隊列消息（曬單數據）：");
            var_dump($this->params);
            // 分析曬單數據
            $orderService = new OrderService();
            $res = $orderService->analyseOne($this->params);
            Log::get('AnalyseJob')->debug('監聽到隊列消息（曬單數據）,進行分析：', $res);
        } catch (\Exception $e) {
            Log::get('AnalyseJob')->debug('監聽到隊列消息（曬單數據）,進行分析：' . $e->getMessage());
        }
    }
}

    3、寫一個專門投遞消息的service

     QueueService.php

<?php

declare(strict_types=1);

namespace App\Service\Producer;

use App\Job\AnalyseJob;
use Hyperf\AsyncQueue\Driver\DriverFactory;
use Hyperf\AsyncQueue\Driver\DriverInterface;

class QueueService
{
    /**
     * @var DriverInterface
     */
    protected $driver;

    public function __construct(DriverFactory $driverFactory)
    {
        $this->driver = $driverFactory->get('default');
    }

    /**
     * 生產消息
     * @param $params
     * @param int $delay
     * @return bool
     */
    public function push($params, int $delay = 0): bool
    {
        return $this->driver->push(new AnalyseJob($params), $delay);
    }
}

    4、投遞消息

    業務中調用service，來投遞消息

<?php

declare(strict_types=1);

namespace App\Controller;

use App\Service\QueueService;
use Hyperf\Di\Annotation\Inject;
use Hyperf\HttpServer\Annotation\AutoController;

/**
 * @AutoController
 */
class UserOrderController extends AbstractController
{
    /**
     * @Inject
     * @var QueueService
     */
    protected $service;

    /**
     * 注解模式投遞消息
     */
    public function example()
    {
        // 創建/編輯成功后，推送隊列消息，消費曬單消息用於比對原始訂單
        if ($ret['code'] == 0) {
            $userOrderRes = $this->orderService->getUserOrderNativeInfo($ret['data']);
            $this->queueService->push($userOrderRes['data']);
        }

        // ...

    }
}

 

參考鏈接：

https://hyperf.wiki/2.1/#/zh-cn/async-queue