震坤行

server/app/Console/Commands/ZhenkhJobCommand.php

@@ -3,6 +3,7 @@ namespace App\Console\Commands;
 
 use App\Jobs\WuxiLabJob;
 use App\Jobs\ZhenkhJob;
+use App\Services\ZhenKhService;
 use Illuminate\Console\Command;
 use Illuminate\Console\ConfirmableTrait;
 
@@ -57,8 +58,41 @@ class ZhenkhJobCommand extends Command
                 exit;
             }
 
-            $productUpdateJob = (new ZhenkhJob($action_type,$params))->delay(100)->onQueue('slow');
-            app('Illuminate\Contracts\Bus\Dispatcher')->dispatch($productUpdateJob);
+            $service = app(ZhenKhService::class);
+            switch($action_type){
+                case 'test' :
+                    $service->test();
+                    break;
+                case 'getUpdateGoods' :
+                    $service->getUpdateSupplierGoods();
+                    break;
+                case 'getUpdateOrders' :
+                    $service->getUpdateSupplierOrders();
+                    break;
+                case 'sendDeliveryOrder' :
+                    $service->sendDeliveryOrder($params);
+                    break;
+                case 'getDeliveryOrderDetail' :
+                    $service->getDeliveryOrderPdf($params);
+                    break;
+                case 'orderNoRollBack' :
+                    $service->orderNoRollBack();
+                    break;
+                case 'createJwtToken' :
+                    $service->createJwtToken();
+                    break;
+                case 'hMUpdateOrdersList' :
+                    $service->hMUpdateOrdersList();
+                    break;
+                case 'sendMail' :
+                    $service->sendMail('订单确认邮件', '订单号:未确认成功');
+                case 'orderConfirm' :
+                    $service->orderConfirm($params);
+                default:
+            }
+
+            /*$productUpdateJob = (new ZhenkhJob($action_type,$params))->delay(100);
+            app('Illuminate\Contracts\Bus\Dispatcher')->dispatch($productUpdateJob);*/
         }
     }
 

server/app/Jobs/queue/ZhenkhQueueJob.php

+<?php
+namespace App\Jobs\queue;
+
+use App\Jobs\Job;
+
+class ZhenkhQueueJob extends Job
+{
+    public $params;
+    /**
+     * Create a new job instance.
+     *
+     * @return void
+     */
+    public function __construct($params= '')
+    {
+        $this->params = $params;
+    }
+
+    /**
+     * Execute the job.
+     *
+     * @return void
+     */
+    public function handle()
+    {
+        if(!empty($this->params)){
+            if(isset($this->params['method']) && method_exists($this->params['consumer'],$this->params['method'])){
+                //app($this->params['consumer'])->$this->params['method']($this->params['params']);
+                $method = $this->params['method'];
+                call_user_func([app($this->params['consumer']),$method],$this->params['params']);
+                //app($this->params['consumer'])->{$method}($this->params['params']);
+            }
+        }
+    }
+}
\ No newline at end of file

server/app/Services/ZhenKhService.php

@@ -2,6 +2,7 @@
 
 namespace App\Services;
 
+use App\Jobs\queue\ZhenkhQueueJob;
 use App\Mailer\MailService;
 use App\Mailer\MailTempl;
 use App\Repositories\Contracts\ThirdApiPlatformRepository;
@@ -126,7 +127,7 @@ class ZhenKhService
                             'consumer' => __CLASS__,
                             'method' => 'consumerBatchUpdateGoodsList'
                         ];
-                        SimpleKafka::produerSend($sendContent,'zkh');
+                        $this->pushQueue($sendContent);
                     }
                 }
 
@@ -175,11 +176,11 @@ class ZhenKhService
                 if($orderList && count($orderList['list']) > 0){
                     foreach($orderList['list']['data'] as $order){
                         if($order['orderStatus'] == 0){
-                            SimpleKafka::produerSend([
+                            $this->pushQueue([
                                 'params' => ['orders' => $order],
                                 'consumer' => __CLASS__,
                                 'method' => 'consumerBatchUpdateOrdersList'
-                            ],'zkh');
+                            ]);
                         }
                     }
                 }
@@ -545,4 +546,15 @@ 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');
+        }
+    }
+
 }

server/bootstrap/app.php

@@ -140,7 +140,7 @@ $app->register(Dingo\Api\Provider\LumenServiceProvider::class);
 
 $app->register(\Jiannei\Enum\Laravel\Providers\LumenServiceProvider::class);
 $app->register(\Jiannei\Response\Laravel\Providers\LumenServiceProvider::class);
-$app->register(\Jiannei\Logger\Laravel\Providers\ServiceProvider::class);
+//$app->register(\Jiannei\Logger\Laravel\Providers\ServiceProvider::class);
 
 $app->register(Maatwebsite\Excel\ExcelServiceProvider::class);
 $app->register(Spatie\CronlessSchedule\CronlessScheduleServiceProvider::class);

server/database/migrations/2022_09_15_133802_create_consumer_queue_jobs_table.php

+<?php
+
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\Schema;
+
+class CreateConsumerQueueJobsTable extends Migration
+{
+    /**
+     * Run the migrations.
+     *
+     * @return void
+     */
+    public function up()
+    {
+        Schema::create('consumer_queue_jobs', function (Blueprint $table) {
+            $table->bigIncrements('id');
+            $table->string('queue')->index();
+            $table->longText('payload');
+            $table->unsignedTinyInteger('attempts');
+            $table->unsignedInteger('reserved_at')->nullable();
+            $table->unsignedInteger('available_at');
+            $table->unsignedInteger('created_at');
+        });
+    }
+
+    /**
+     * Reverse the migrations.
+     *
+     * @return void
+     */
+    public function down()
+    {
+        Schema::dropIfExists('consumer_queue_jobs');
+    }
+}