队列更新

server/app/Console/Commands/BideJobCommand.php

@@ -2,6 +2,7 @@
 namespace App\Console\Commands;
 
 use App\Jobs\BideJob;
+use App\Services\BideService;
 use Illuminate\Console\Command;
 use Illuminate\Console\ConfirmableTrait;
 
@@ -56,8 +57,15 @@ class BideJobCommand extends Command
                 exit;
             }
 
-            $productUpdateJob = (new BideJob($action_type,$params))->delay(100)->onQueue('slow');
-            app('Illuminate\Contracts\Bus\Dispatcher')->dispatch($productUpdateJob);
+            $service = app(BideService::class);
+            switch($action_type){
+                case 'batchUpdateProduct' :
+                default:
+                    $service->batchUpdateProducts($params);
+                    break;
+            }
+            //$productUpdateJob = (new BideJob($action_type,$params))->delay(100)->onQueue('slow');
+            //app('Illuminate\Contracts\Bus\Dispatcher')->dispatch($productUpdateJob);
         }
     }
 

server/app/Console/Commands/BjsJobCommand.php

@@ -2,6 +2,7 @@
 namespace App\Console\Commands;
 
 use App\Jobs\BjsJob;
+use App\Services\BjsService;
 use Illuminate\Console\Command;
 use Illuminate\Console\ConfirmableTrait;
 
@@ -56,8 +57,16 @@ class BjsJobCommand extends Command
                 exit;
             }
 
-            $productUpdateJob = (new BjsJob($action_type,$params))->delay(100)->onQueue('slow');
-            app('Illuminate\Contracts\Bus\Dispatcher')->dispatch($productUpdateJob);
+            $service = app(BjsService::class);
+            switch($action_type){
+                case 'batchUpdateProduct' :
+                default:
+                    $service->batchUpdateProducts($params);
+                    break;
+            }
+
+            //$productUpdateJob = (new BjsJob($action_type,$params))->delay(100)->onQueue('slow');
+            //app('Illuminate\Contracts\Bus\Dispatcher')->dispatch($productUpdateJob);
         }
     }
 

server/app/Console/Commands/IntegleJobCommand.php

@@ -10,6 +10,7 @@ namespace App\Console\Commands;
 
 use App\Jobs\InteglePlatformJob;
 use App\Jobs\WuxiLabJob;
+use App\Services\InteglePlatformService;
 use Illuminate\Console\Command;
 use Illuminate\Console\ConfirmableTrait;
 
@@ -62,8 +63,17 @@ class IntegleJobCommand extends Command
                 $this->error('缺少命令参数,请输入具体的参数命令.如需帮助请输入 --help');
                 exit;
             }
-            $productUpdateJob = (new InteglePlatformJob($action_type,$params))->delay(100)->onQueue('slow');
-            app('Illuminate\Contracts\Bus\Dispatcher')->dispatch($productUpdateJob);
+
+            $service = app(InteglePlatformService::class);
+            switch($action_type){
+                case 'batchUpdateProduct' :
+                    $service->batchUpdateProducts();
+                    break;
+                default:
+            }
+
+            //$productUpdateJob = (new InteglePlatformJob($action_type,$params))->delay(100)->onQueue('slow');
+            //app('Illuminate\Contracts\Bus\Dispatcher')->dispatch($productUpdateJob);
         }
     }
 

server/app/Jobs/queue/ZhenkhQueueJob.php → server/app/Jobs/queue/QueueJob.php

@@ -3,7 +3,7 @@ namespace App\Jobs\queue;
 
 use App\Jobs\Job;
 
-class ZhenkhQueueJob extends Job
+class QueueJob extends Job
 {
     public $params;
     /**

server/app/Services/BideService.php

@@ -11,6 +11,7 @@
 
 namespace App\Services;
 
+use App\Jobs\queue\QueueJob;
 use App\Repositories\Contracts\ThirdApiPlatformRepository;
 use App\Services\Kafka\KafkaService;
 use App\Services\ThirdPlatform\Api\BideApiService;
@@ -33,14 +34,12 @@ class BideService
         // yield 循环逐行读取csv内容
         while ($generator->valid()) {
             $content = $generator->current();
-            app(KafkaService::class)->produerSend(
-                [
-                    'params' => ['cas' => current($content)],
-                    'consumer' => __CLASS__,
-                    'method' => 'batchUpdateApi'
-
-                ]
-            );
+            $this->apiService->pushQueue([
+                'params' => ['cas' => current($content)],
+                'consumer' => __CLASS__,
+                'method' => 'batchUpdateApi'
+
+            ],'bide');
             $generator->next();
         }
         //$generator->rewind();

server/app/Services/BjsService.php

@@ -33,14 +33,12 @@ class BjsService
         // yield 循环逐行读取csv内容
         while ($generator->valid()) {
             $content = $generator->current();
-            app(KafkaService::class)->produerSend(
-                [
-                    'params' => ['cas' => current($content)],
-                    'consumer' => __CLASS__,
-                    'method' => 'batchUpdateApi'
-
-                ]
-            );
+            $this->apiService->pushQueue([
+                'params' => ['cas' => current($content)],
+                'consumer' => __CLASS__,
+                'method' => 'batchUpdateApi'
+
+            ],'bjs');
             $generator->next();
         }
         //$generator->rewind();

server/app/Services/InteglePlatformService.php

@@ -45,13 +45,11 @@ class InteglePlatformService
                     }
                 }
                 if(!empty($rawList)){
-                    SimpleKafka::produerSend(
-                        [
-                            'params' => ['rawList' => $rawList,'packageList' => $newPackages],
-                            'consumer' => 'App\Services\InteglePlatformService',
-                            'method' => 'batchUpdateApi'
-                        ]
-                    );
+                    $this->apiService->pushQueue([
+                        'params' => ['rawList' => $rawList,'packageList' => $newPackages],
+                        'consumer' => 'App\Services\InteglePlatformService',
+                        'method' => 'batchUpdateApi'
+                    ],'integle');
                     $page ++;
                 }
             }
@@ -63,13 +61,11 @@ class InteglePlatformService
         $result = $this->apiService->pushBatchUpdateProduct($updateData);
         if($result){
             if($result['status'] === false){
-                SimpleKafka::produerSend(
-                    [
-                        'consumer' => 'App\Services\InteglePlatformService',
-                        'method' => 'batchUpdateApi',
-                        'params' => $updateData
-                    ]
-                );
+                $this->apiService->pushQueue([
+                    'consumer' => 'App\Services\InteglePlatformService',
+                    'method' => 'batchUpdateApi',
+                    'params' => $updateData
+                ],'integle');
             }
         }
         unset($result['status']);

server/app/Services/ThirdPlatform/PlatformAbstractService.php

@@ -11,9 +11,11 @@
 
 namespace App\Services\ThirdPlatform;
 
+use App\Jobs\queue\QueueJob;
 use App\Repositories\Contracts\ThirdApiPlatformRepository;
 use App\Services\Kafka\KafkaService;
 use App\Services\PlatformDataEntriesService;
+use App\Support\Facades\SimpleKafka;
 
 abstract class PlatformAbstractService
 {
@@ -114,4 +116,15 @@ abstract class PlatformAbstractService
     {
         return $this->dataEntriesService->removeEntriesValues($this->getPlatformInfo('id'),$dataKey);
     }
+
+    public function pushQueue($pushContent, $queueName)
+    {
+        if(env('QUEUE_CONNECTION') == 'database'){
+            app('Illuminate\Contracts\Bus\Dispatcher')->dispatch(
+                (new QueueJob($pushContent))->onQueue($queueName)
+            );
+        }else{
+            SimpleKafka::produerSend($pushContent,$queueName);
+        }
+    }
 }

server/app/Services/ZhenKhService.php

@@ -2,7 +2,7 @@
 
 namespace App\Services;
 
-use App\Jobs\queue\ZhenkhQueueJob;
+use App\Jobs\queue\QueueJob;
 use App\Mailer\MailService;
 use App\Mailer\MailTempl;
 use App\Repositories\Contracts\ThirdApiPlatformRepository;
@@ -127,7 +127,7 @@ class ZhenKhService
                             'consumer' => __CLASS__,
                             'method' => 'consumerBatchUpdateGoodsList'
                         ];
-                        $this->pushQueue($sendContent);
+                        $this->apiService->pushQueue($sendContent,'zkh');
                     }
                 }
 
@@ -176,11 +176,11 @@ class ZhenKhService
                 if($orderList && count($orderList['list']) > 0){
                     foreach($orderList['list']['data'] as $order){
                         if($order['orderStatus'] == 0){
-                            $this->pushQueue([
+                            $this->apiService->pushQueue([
                                 'params' => ['orders' => $order],
                                 'consumer' => __CLASS__,
                                 'method' => 'consumerBatchUpdateOrdersList'
-                            ]);
+                            ],'zkh');
                         }
                     }
                 }
@@ -546,15 +546,4 @@ class ZhenKhService
         }
     }
 
-    private function pushQueue($pushContent)
-    {
-        if(env('QUEUE_CONNECTION') == 'database'){
-            app('Illuminate\Contracts\Bus\Dispatcher')->dispatch(
-                (new ZhenkhQueueJob($pushContent))->onQueue('zkh')
-            );
-        }else{
-            SimpleKafka::produerSend($pushContent,'zkh');
-        }
-    }
-
 }