hcsl/app/Jobs/ProcessDataJob.php

<?php

namespace App\Jobs;

use App\Model\Box;
use App\Model\DispatchSub;
use App\Model\ErrorTable;
use App\Service\FinishedOrderService;
use App\Service\FyyOrderService;
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;

    //队列名称
    const job_one = 'finished_operation';//完工
    const job_two = 'box_operation';//包装

    protected $data;
    protected $user;
    protected $type;

    public $tries = 0;
    public $timeout = 60;

    //1 代表产成品入库  2 销售出库单  3 代表产成品入库手机端  4 销售出库单
    protected $function = [
        1 => 'U8Rdrecord10Save',
        2 => 'U8Rdrecord32Save',
        3 => 'U8Rdrecord10SaveMobile'
    ];

    //数据回退 标记了的单据数据状态改为0
    protected $function_reback = [
        1 => 'reBackOne',
        2 => 'reBackTwo',
        3 => 'reBackThree',
    ];

    protected $jobs = [
        1 => self::job_one,
        2 => self::job_two,
        3 => self::job_one,
    ];

    /**
     * Create a new job instance.
     *  $data = [
            'result' => $msg, //查询结果
            'data' => $data, //用户提交的参数
        ];
     * $user 提交用户的信息
     * $type //1 代表产成品入库（完工操作）  2 销售出库单（包装单扫描出库）
     *
     * @return void
     */
    public function __construct($data, $user = [], $type)
    {
        $this->data = $data;
        $this->user = $user ?? [];
        $this->type = $type;
    }

    /**
     *
     * file_put_contents('charge.txt',"标记位置退出".PHP_EOL,8);
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    {
        try {
            $function = $this->function[$this->type] ?? '';//调用方法 入库
            $function_back = $this->function_reback[$this->type] ?? '';//数据回退方法
            if(empty($function))  return;

            list($status,$msg) = $this->$function();

            if(! $status) $this->errorSettle($msg);
        } catch (\Exception $e) {
            $this->$function_back();
            $this->recordErrorTable($e->getFile() . $e->getMessage() . $e->getLine());
        }

        //输出信息
        $this->echoMessage(new ConsoleOutput());
    }

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

    //产成品入库
    private function U8Rdrecord10Save(){
        $service = new FinishedOrderService();
        //标记
        list($status,$msg) = $service->addInJob($this->data['result'],$this->data['data'],$this->user);
        return [$status,$msg];
    }

    //产成品入库手机端
    private function U8Rdrecord10SaveMobile(){
        $service = new FinishedOrderService();
        //标记
        list($status,$msg) = $service->addMobileInJob($this->data['result'],$this->data['data'],$this->data['quantity_count'],$this->user);
        return [$status,$msg];
    }

    //产成品入库 相关数据回退
    private function reBackOne(){
        //数据回退
        $data = $this->data['data'];

        $database = $this->settleDatabase();

        // 连接到指定数据库连接
        DB::connection($database)->table('dispatch_sub')
            ->whereIn('id',$data['id'])
            ->update(['job_status' => 0]);
    }

    //产成品入库 相关数据回退
    private function reBackThree(){
        //数据回退
        $data = $this->data['data'];

        $database = $this->settleDatabase();

        // 连接到指定数据库连接
        DB::connection($database)->table('dispatch_sub')
            ->whereIn('id',array_column($data,'id'))
            ->update(['job_status' => 0]);
    }

    //销售单出库
    private function U8Rdrecord32Save(){
        $service = new FyyOrderService();
        list($status,$msg) = $service->addInJob($this->data['result'],$this->data['data'],$this->user);
        return [$status,$msg];
    }

    //销售单出库 相关数据回退
    private function reBackTwo(){
        //数据回退
        $data = $this->data['data'];

        $database = $this->settleDatabase();

        // 连接到指定数据库连接
        DB::connection($database)->table('box')
            ->whereIn('order_no',$data['order_no'])
            ->update(['state' => 0]);
    }

    private function errorSettle($msg){
        $redis = Redis::connection();

        $order_failures_key = md5(json_encode($this->data));

        // 从Redis中获取失败计数
        $failureCount = $redis->hIncrBy('order_failures', $order_failures_key, 1);

        //队列名
        $job = $this->jobs[$this->type];

        if ($failureCount < 1) {
            // 将任务重新放回队列
            self::dispatch($this->data,$this->user,$this->type)->onQueue($job)->delay(now()->addSeconds(2));
        } else {
            // 删除失败计数
            $redis->hDel('order_failures', $order_failures_key);

            //数据回退
            $function_back = $this->function_reback[$this->type] ?? '';
            $this->$function_back();

            //记录错误
            $this->recordErrorTable($msg);
        }
    }

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

        // 连接到指定数据库连接
        DB::connection($database)->table('error_table')->insert([
            'msg' => $msg,
            'data' => json_encode($this->data),
            'user_id' => $this->user['id'],
            'user_operation_time' => $this->user['operate_time'],
            'type' => $this->type
        ]);
    }

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

    public function settleDatabase(){
        $zt = $this->user['zt'] ?? '';

        return (new FinishedOrderService())->getConnectionName($zt);
    }
}