xyzx/app/Jobs/AssetDeviceJob.php

<?php

namespace App\Jobs;

use App\Model\Asset;
use App\Model\AssetTemp;
use App\Service\InOutOptionService;
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 Symfony\Component\Console\Output\ConsoleOutput;
use Symfony\Component\Console\Output\OutputInterface;

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

    protected $data;
    public $timeout = 1800;

    /**
     * Create a new job instance
     *
     * @return void
     */
    public function __construct($data)
    {
        $this->data = $data;
    }

    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    {
        try{
            DB::beginTransaction();

            $this->settle();

            DB::commit();
            //输出信息 测试
            $this->echoMessage(new ConsoleOutput());
        }catch (\Exception $exception){
            DB::rollBack();
            file_put_contents('record_error.txt',date("Y-m-d H:i:s",time()).json_encode($this->data) . PHP_EOL.$exception->getMessage()."line" . $exception->getLine().PHP_EOL,8);
        }
    }

    public function settle(){
        $data = $this->data;
        if(empty($data)){
            file_put_contents('record_data.txt',date("Y-m-d H:i:s",time())."原数据：".json_encode($data).PHP_EOL,8);
            return;
        }

        if(empty($data['assetOas'])) {
            file_put_contents('record_data.txt',date("Y-m-d H:i:s",time())."数据结构错误".PHP_EOL,8);
            return;
        }

        //内存设置
        ini_set('memory_limit', -1);

        //软删除
        $time = time();
        Asset::where('del_time',0)
            ->where('crt_time','<', strtotime(date('Y-m-d 00:00:00')))
            ->update(['del_time' => $time]);

        $this->assetTempInsert();
        echo 'TEMP  OVER-------------------'. PHP_EOL;;

        $this->assetInsert();
        echo 'OVER-------------------';

        AssetTemp::truncate();
    }

    //临时表数据插入
    public function assetTempInsert(){
        $data = $this->data;

        $batchSize = 100;
        $totalAssets = count($data['assetOas']);
        for ($i = 0; $i < $totalAssets; $i += $batchSize) {
            $batch = array_slice($data['assetOas'], $i, $batchSize);

            $insert = [];

            // 对每个批次的数据进行处理
            foreach ($batch as $value) {
                if(empty($value['singleCode'])) continue;

                $insert[] = [
                    "assetCode" => $value['assetCode'] ?? "",
                    "assetNo" => $value['assetNo'] ?? "",
                    "assetType" => $value['assetType'] ?? "",
                    "brand" => $value['brand'] ?? "",
                    "expectedLife" => $value['expectedLife'] ?? "",
                    "gs1" => $value['gs1'] ?? "",
                    "isKey" => $value['isKey'] ?? "",
                    "kind" => $value['kind'] ?? "",
                    "located" => $value['located'] ?? "",
                    "name" => $value['name'] ?? "",
                    "nextCalibrationTime" => $value['nextCalibrationTime'] ?? "",
                    "originalValue" => $value['originalValue'] ?? "",
                    "purchaseTime" => $value['purchaseTime'] ?? "",
                    "remark" => $value['remark'] ?? "",
                    "singleCode" => $value['singleCode'] ?? "",
                    "startUseDate" => $value['startUseDate'] ?? "",
                    "type" => $value['type'] ?? "",
                    "useDept" => $value['useDept'] ?? "",
                    "userName" => $value['userName'] ?? "",
                    "version" => $value['version'] ?? "",
                ];
            }

            // 在处理完每个批次后，进行一些内存清理操作
            unset($batch);
            if(! empty($insert)) {
                echo "insert Temp " . PHP_EOL;
                AssetTemp::insert($insert);
            }
        }
    }

    //正式数据插入或更新
    public function assetInsert(){
        DB::table('asset_temp')
            ->select('*')
            ->orderBy('id','desc')
            ->chunk(500,function ($data){
                $data_array = $data->toArray();

                $time = time();
                $update = $insert = [];
                foreach ($data_array as $value){
                    $bool = Asset::where('singleCode',$value->singleCode)->exists();
                    if($bool){
                        $update[] = [
                            "assetCode" => $value->assetCode ?? "",
                            "assetNo" => $value->assetNo ?? "",
                            "assetType" => $value->assetType ?? "",
                            "brand" => $value->brand ?? "",
                            "expectedLife" => $value->expectedLife ?? "",
                            "gs1" => $value->gs1 ?? "",
                            "isKey" => $value->isKey ?? "",
                            "kind" => $value->kind ?? "",
                            "located" => $value->located ?? "",
                            "name" => $value->name ?? "",
                            "nextCalibrationTime" => $value->nextCalibrationTime ?? "",
                            "originalValue" => $value->originalValue ?? "",
                            "purchaseTime" => $value->purchaseTime ?? "",
                            "remark" => $value->remark ?? "",
                            "singleCode" => $value->singleCode ?? "",
                            "startUseDate" => $value->startUseDate ?? "",
                            "type" => $value->type ?? "",
                            "useDept" => $value->useDept ?? "",
                            "userName" => $value->userName ?? "",
                            "version" => $value->version ?? "",
                            "del_time" => 0,
                        ];
                    }else{
                        $insert[] = [
                            "assetCode" => $value->assetCode ?? "",
                            "assetNo" => $value->assetNo ?? "",
                            "assetType" => $value->assetType ?? "",
                            "brand" => $value->brand ?? "",
                            "expectedLife" => $value->expectedLife ?? "",
                            "gs1" => $value->gs1 ?? "",
                            "isKey" => $value->isKey ?? "",
                            "kind" => $value->kind ?? "",
                            "located" => $value->located ?? "",
                            "name" => $value->name ?? "",
                            "nextCalibrationTime" => $value->nextCalibrationTime ?? "",
                            "originalValue" => $value->originalValue ?? "",
                            "purchaseTime" => $value->purchaseTime ?? "",
                            "remark" => $value->remark ?? "",
                            "singleCode" => $value->singleCode ?? "",
                            "startUseDate" => $value->startUseDate ?? "",
                            "type" => $value->type ?? "",
                            "useDept" => $value->useDept ?? "",
                            "userName" => $value->userName ?? "",
                            "version" => $value->version ?? "",
                            "crt_time" => $time,
                        ];
                    }
                }
                if(! empty($update)){
                    foreach ($update as $value){
                        Asset::where('singleCode',$value['singleCode'])->update($value);
                    }
                }
                if(! empty($insert)) Asset::insert($insert);

                echo "insert OR update ing" . PHP_EOL;
            });
    }

    protected function echoMessage(OutputInterface $output)
    {
        $output->writeln($this->data);
    }
}