feat: 增加队列并行消费能力

Author: peter

README.md

@@ -20,9 +20,14 @@
 本项目设计时考虑了多商户多店铺的情况，所以大部分表都需要有`merchant_id`与`merchant_store_id`字段。项目初始化时请把相应配置填入`amazon_app`表中(**后续会调整该表结构，把不同地区的refresh_token放到amazon_app_region中**)。
 
 
-
 ### 常用命令
 ```
+# 刷新Token缓存
+
+crontab:amazon:refresh-app-token
+
+
+
 # 创建报告
 ### 强制创建销售与流量报告， 时间范围为2023-12-01到2023-12-20，循环创建每一天的报告
 > php bin/hyperf.php amazon:report:create 1 1 GET_SALES_AND_TRAFFIC_REPORT --report_start_date=2023-12-01 --report_end_date=2023-12-20 --is_range_date=1 --is_force_create=1
@@ -34,9 +39,9 @@
 > php bin/hyperf.php amazon:report:action
 
 # 获取周期报告
->
+> php bin/hyperf.php amazon:report:gets 1 1 us-east-1 GET_DATE_RANGE_FINANCIAL_TRANSACTION_DATA
 # 拉取周期报告
->
+> 
 # 处理周期报告
 >
 

app/Command/Amazon/Report/ReportGetDocument.php

@@ -39,6 +39,6 @@ public function configure(): void
      */
     public function handle(): void
     {
-        (new AmazonGetReportDocumentQueue())->pop();
+        (new AmazonGetReportDocumentQueue())->coPop(30);//并行请求的数量
     }
 }

app/Command/Crontab/Amazon/RefreshAppToken.php

@@ -15,6 +15,7 @@
 use App\Model\AmazonAppRegionModel;
 use App\Util\AmazonApp;
 use App\Util\AmazonSDK;
+use App\Util\Constants;
 use Hyperf\Command\Annotation\Command;
 use Hyperf\Command\Command as HyperfCommand;
 use Hyperf\Context\ApplicationContext;
@@ -56,6 +57,7 @@ public function handle(): void
             $amazonAppRegionCollections = AmazonAppRegionModel::query()
                 ->where('merchant_id', $merchant_id)
                 ->where('merchant_store_id', $merchant_store_id)
+                ->where('status', Constants::STATUS_ACTIVE)
                 ->get();
             if ($amazonAppRegionCollections->isEmpty()) {
                 return false;

app/Command/Crontab/Schedules.php

@@ -28,9 +28,6 @@ public function configure(): void
     public function handle(): void
     {
 
-        set_time_limit(0);
-        ini_set('memory_limit', '1024M');
-
         $console = ApplicationContext::getContainer()->get(ConsoleLog::class);
 
         // 指令输出

app/Queue/AbstractQueue.php

@@ -43,6 +43,8 @@ abstract public function push(QueueDataInterface $queueData): bool;
 
     abstract public function pop(): bool;
 
+    abstract public function coPop(): bool;
+
     abstract public function handleQueueData(QueueDataInterface $queueData): bool;
 
     abstract public function len(): int;

app/Queue/AmazonGetReportDocumentQueue.php

@@ -103,7 +103,7 @@ public function handleQueueData(QueueDataInterface $queueData): bool
                 })();
             } elseif (file_exists($file_path)) {
                 $log = sprintf('Get Document 报告已存在，直接进入队列. report_type: %s  report_document_id: %s  file_path:%s merchant_id: %s merchant_store_id: %s', $report_type, $report_document_id, $file_path, $merchant_id, $merchant_store_id);
-                $console->highlight($log);
+                $console->warning($log);
                 goto end;
             } else {
                 $retry = 10;

app/Queue/Queue.php

@@ -13,8 +13,11 @@
 use App\Queue\Data\QueueData;
 use App\Queue\Data\QueueDataInterface;
 use App\Util\Log\QueueLog;
+use Hyperf\Collection\Collection;
 use Hyperf\Context\ApplicationContext;
 use Hyperf\Contract\StdoutLoggerInterface;
+use Hyperf\Coroutine\Parallel;
+use Hyperf\Engine\Coroutine;
 use Psr\Container\ContainerExceptionInterface;
 use Psr\Container\NotFoundExceptionInterface;
 
@@ -64,11 +67,10 @@ public function pop(): bool
         $timeout = $this->timeout;
 
         $retryInterval = $this->retryInterval; // 消息重试次数
-
         while (true) {
             try {
                 $pop = $this->redis->brpop($this->queue_name, $timeout);
-                if (is_array($pop)) {
+                if (empty($pop)) {
                     pcntl_signal_dispatch();
                     $console->info(sprintf('进程[%s] pid:%s 队列为空，自动退出', cli_get_process_title(), $pid));
                     break;
@@ -124,6 +126,93 @@ public function pop(): bool
         return true;
     }
 
+    public function coPop(int $parallel_num = 100): bool
+    {
+
+        $console = ApplicationContext::getContainer()->get(StdoutLoggerInterface::class);
+        $logger = ApplicationContext::getContainer()->get(QueueLog::class);
+
+        if ($parallel_num < 1) {
+            $console->error(sprintf('队列:%s 并行消费数量不能小于1。当前并行数量为:%s', $this->queue_name, $parallel_num));
+            return false;
+        }
+
+        $pid = posix_getpid();
+
+        $process_title = $this->queue_name . '-' . $pid;
+        cli_set_process_title($process_title);
+
+        $timeout = $this->timeout;
+
+        $retryInterval = $this->retryInterval; // 消息重试次数
+
+        while ($this->len()) {
+            $collections = new Collection();
+
+            while (true) {
+                try {
+                    $pop = $this->redis->brpop($this->queue_name, $timeout);
+                    if (empty($pop)) {
+                        break;
+                    }
+                    $collections->push($pop[1]);
+                } catch (\RedisException $exception) {
+                    $logger->error(sprintf('队列：%s 连接Redis异常.%s', $this->queue_name, $exception->getMessage()));
+                    break;
+                }
+
+                if ($collections->count() >= $parallel_num) {
+                    break;
+                }
+            }
+
+            $count = $collections->count();
+
+            $console->notice(sprintf('进程[%s] pid:%s 消费数据长度%s', cli_get_process_title(), $pid, $count));
+            if ($count === 0) {
+                return true;
+            }
+
+            $class = $this->getQueueDataClass();
+
+            $parallel = new Parallel($parallel_num);
+
+            $that = $this;
+            /**
+             * @var QueueData $dataObject
+             */
+            $collections->each(function ($item) use ($that, $class, $parallel, $logger, $console) {
+                /**
+                 * @var QueueData $dataObject
+                 */
+                $dataObject = new $class();
+
+                $arr = $dataObject->toArr($item);
+                $dataObject->parse($arr);
+
+                $parallel->add(function () use ($that, $dataObject, $item, $console) {
+
+                    $t1 = microtime(true);
+
+                    $handle = $that->handleQueueData($dataObject);
+                    if ($handle === false) {
+                    }
+
+                    $t2 = microtime(true);
+
+                    $co_id = Coroutine::id();//当前协程id
+
+                    $console->info(sprintf('队列：%s 消费数据. data:%s  耗时:%s 秒 co_id:%s', $this->queue_name, json_encode($item, JSON_THROW_ON_ERROR), round($t2 - $t1, 3), $co_id));
+
+                    return $co_id;
+                });
+            });
+
+            $results = $parallel->wait();
+        }
+        return true;
+    }
+
     /**
      * @throws \RedisException
      */

app/Util/AmazonApp.php

@@ -94,6 +94,7 @@ public static function tok(int $merchant_id, int $merchant_store_id, callable $f
             $amazonAppRegionCollections = AmazonAppRegionModel::query()
                 ->where('merchant_id', $merchant_id)
                 ->where('merchant_store_id', $merchant_store_id)
+                ->where('status', Constants::STATUS_ACTIVE)
                 ->get();
             if ($amazonAppRegionCollections->isEmpty()) {
                 return false;
@@ -170,6 +171,7 @@ public static function tok2(int $merchant_id, int $merchant_store_id, string $re
                     ->where('merchant_id', $merchant_id)
                     ->where('merchant_store_id', $merchant_store_id)
                     ->where('region', $region)
+                    ->where('status', Constants::STATUS_ACTIVE)
                     ->firstOrFail();
             } catch (ModelNotFoundException $modelNotFoundException) {
                 $log = sprintf('Amazon App SDK构建失败，AmazonAppRegion查询失败. 错误信息:%s merchant_id:%s merchant_store_id:%s region:%s ', $modelNotFoundException->getMessage(), $merchant_id, $merchant_store_id, $region);

app/Util/AmazonSDK.php

@@ -272,7 +272,7 @@ public function getRefreshToken(): string
     public function getSdk(string $region, bool $force_refresh = false): SellingPartnerSDK
     {
         $factory = new Psr17Factory();
-        $client = new Curl($factory);
+        $client = new Curl($factory, ['timeout' => 10]);// 超时时间 单位：秒
 
         $sts = new STSClient(
             $client,

config/autoload/databases.php

@@ -30,9 +30,9 @@
         'prefix' => env('DB_PREFIX', ''),
         'pool' => [
             'min_connections' => 1,
-            'max_connections' => 10,
+            'max_connections' => 50,
             'connect_timeout' => 10.0,
-            'wait_timeout' => 3.0,
+            'wait_timeout' => 10.0,
             'heartbeat' => -1,
             'max_idle_time' => (float) env('DB_MAX_IDLE_TIME', 60),
         ],