drb/app/Jobs/ProcessDataJob.php

<?php

namespace App\Jobs;

use App\Model\ErrorTable;
use App\Model\U8Job;
use App\Service\ProductService;
use App\Service\TPlusDatabaseServerService;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Redis;
use MongoDB\Driver\Exception\Exception;
use Symfony\Component\Console\Output\ConsoleOutput;
use Symfony\Component\Console\Output\OutputInterface;

class ProcessDataJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $data;

    public $tries = 1;
    public $timeout = 60;

    //1 采购  2 销售（合同） 3 总社库存
    protected $function = [
        1 => 'U8PO_PomainSave',
        2 => 'U8SaleOrderSave',
        3 => 'U8StockToMy',
    ];

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

    public function handle()
    {
        try {
            $function = $this->function[$this->data['type']] ?? '';
            if(empty($function)) return;

            //调用同步方法
            $this->$function();
        } catch (\Throwable $e) {
            $this->delete();
        }
    }

    //采购
    private function U8PO_PomainSave(){
        $service = new TPlusDatabaseServerService();
        if(! empty($service->error)) {
            $service->finalSettle($this->data['id'], U8Job::one, $service->error);
            return;
        }

        $service->U8PO_PomainSave($this->data['id'], $this->data['user_name']);
    }

    //销售（合同）
    private function U8SaleOrderSave(){
        $service = new TPlusDatabaseServerService();
        if(! empty($service->error)) {
            $service->finalSettle($this->data['id'], U8Job::two, $service->error);
            return;
        }

        $service->U8SaleOrderSave($this->data['id'], $this->data['user_name']);
    }

//    public function failed($exception)
//    {
//        // 记录失败错误信息到日志或其他媒介
//        $errorMessage = $exception->getFile() . $exception->getMessage() . $exception->getLine();
//        $this->recordErrorTable($errorMessage);
//    }

    //总社库存
    private function U8StockToMy(){
        $service = new ProductService();
        list($status, $msg) = $service->writeStock($this->data);

        $service = new TPlusDatabaseServerService();
        $service->finalSettle(-1, U8Job::three, $msg);
    }

    private function recordErrorTable($msg){
        $data = $this->data;

        ErrorTable::insert([
            'msg' => $msg,
            'data' => json_encode($this->data),
            'user_id' => $data['user']['id'],
            'user_operation_time' => $data['user']['operate_time'],
            'type' => $data['type']
        ]);
    }

    protected function echoMessage(OutputInterface $output)
    {
        //输出消息
        $output->writeln(json_encode($this->data));
    }
}