baoshi-was/app/Jobs/CacheShelfTaskJob.php

<?php

namespace App\Jobs;

use App\Components\ErrorPush;
use App\Services\ForeignHaiRoboticsService;
use App\Services\LogService;
use App\Station;
use App\StationTaskMaterialBox;
use App\TaskTransaction;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Database\Eloquent\Builder;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;

class CacheShelfTaskJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, ErrorPush;

    protected $key;
    protected $count;
    /**
     * Create a new job instance.
     *
     * @return void
     */
    public function __construct(string $key,int $count)
    {
        $this->count = $count;
        $this->key = $key;
    }

    /**
     * Execute the job.
     *
     * @return void
     * @throws
     */
    public function handle()
    {
        /** @var ForeignHaiRoboticsService $service */
        $service = app("ForeignHaiRoboticsService");
        switch ($this->key){
            case "CACHE_SHELF_AVAILABLE"://缓存架释放呼叫
                //等待一定时间来合并同类请求至此
                $available = Cache::get($this->key,0);
                Log::debug("队列事务1",["当前数量:{$this->count}","缓存数量：$available"]);
                if ($this->count!=$available)return;
                Cache::forget($this->key); //无论是否开始分发 都清除本次缓存架的计数器
                //获取可用缓存架
                $stations = app("StationService")->getCacheShelf(true);
                Log::debug("队列事务2",["可用缓存架:{$stations->count()}"]);
                if ($stations->count()==0)break;
                //检查事务 尝试分发任务 改变下方序列来控制分发顺序 逐级分发 一次成功就终止
                if ($this->dispatchOutTask($stations,$service))break; //首先尝试向出库事务分发 分发成功跳出
                if ($this->dispatchInTask($stations,$service))break;  //尝试向入库事务分发
                break;
            default://入库呼叫
                if (!Cache::has($this->key))return;
                /** @var Collection $task */
                list($task,$location) = Cache::get($this->key);
                if ($this->count!=$task->count())return;
                $dataToPost = $service->makeJson_move_multi($task, '缓存架入立架', $location);
                $controlSuccess = $service->controlHaiRobot($dataToPost,$task,'缓存架入立架');
                $tIds = [];
                $task->each(function ($t)use(&$tIds){$tIds[] = $t->id;});
                StationTaskMaterialBox::query()->whereIn("id",$tIds)
                    ->where("status","待处理")->update(['status' => $controlSuccess ? '处理中' : '异常']);
                Cache::forget($this->key);
                //if ($controlSuccess)$this->materialBoxMappingCacheShelf($task,$location);
        }
    }

    /**
     * 料箱映射缓存架，因为建立的入立架任务源库位是立库，无法获取真实源库位，所以在此拿到映射库位
     * 在料箱被取走时通过任务料箱号获取对应库位，来标记该缓存架库位可以被执行任务了 StationTaskMaterialBoxService:markHasTaken
     * 出库会启用库位占用逻辑  入库分为：人工入库与系统自动入库 人工控制入库由人工自己辨识库位可用度，而系统入库只能通过此映射来拿到可用库位信息
     *
     * @param Collection $task
     * @param Collection $location
     *
     * @return void
     */
    public function materialBoxMappingCacheShelf(Collection $task,Collection $location)
    {
        $map = Cache::get("CACHE_SHELF_MAPPING",function (){return [];});
        foreach ($task as $key=>$obj)$map[$obj->material_box_id] = $location[$key];
        Cache::forever("CACHE_SHELF_MAPPING",$map);
    }

    /**
     * 分发出库任务
     *
     * @param $stations
     * @param $service
     * @return bool
     */
    private function dispatchOutTask(&$stations,$service):bool
    {
        DB::beginTransaction();
        try {
            $tasks = TaskTransaction::query()->selectRaw("task_id,GROUP_CONCAT(id) AS ids,id")->with("task")
                ->where("type","出库")->whereHas("task",function ($query){
                    $query->where("status","待处理");
                })->where("status",3)->lockForUpdate()
                ->where("mark",2)->groupBy("task_id")->get(); //检索等待的队列事务来获取对应任务
            Log::debug("队列事务3",["队列出库任务:{$tasks->count()}"]);
            if ($tasks->count()==0)return false;
            if ($this->dispatchTask($tasks,$stations,$service,function ($obj,$stationId,$time,&$updateTransaction){
                if ($obj->ids!=$obj->id){
                    $ids = explode(",",$obj->ids);
                    foreach ($ids as $id)$updateTransaction[] = ["id"=>$id,"to_station_id"=>$stationId,"status"=>0,"updated_at"=>$time];
                }else $updateTransaction[] = ["id"=>$obj->id,"to_station_id"=>$stationId,"status"=>0,"updated_at"=>$time];
            },function ($service,$toLocation,$task,$prefix){
                return $service->fetchGroup_multiLocation($toLocation,$task,$prefix,'立架出至缓存架',20,false);
            },"to_station_id")){
                Log::debug("队列事务4",["缓存架剩余数量：{$stations->count()}"]);
                DB::commit();return $stations->count()==0;} //缓存架用完 跳出，否则接着分发
            DB::rollBack();
        }catch (\Exception $e){
            DB::rollBack();
            $this->push(__METHOD__."->".__LINE__,"出库队列执行错误",$e->getMessage()." | 当前任务数：".$this->count." | 缓存信息：".(isset($available) ? json_encode($available) : ''));
        }
        return false;
    }

    /**
     * 分发入库任务
     *
     * @param $stations
     * @param $service
     * @return bool
     */
    private function dispatchInTask(&$stations,$service):bool
    {
        DB::beginTransaction();
        try {
            $tasks = TaskTransaction::query()->with("task")
                ->where("type","入库")->whereHas("task",function ($query){
                    $query->where("status","待处理");
                })->where("status",3)->lockForUpdate()
                ->where("mark",1)->get(); //检索等待的队列事务来获取对应任务
            Log::debug("队列事务3",["队列入库任务:{$tasks->count()}"]);
            if ($tasks->count()==0)return false;
            if ($this->dispatchTask($tasks,$stations,$service,function ($obj,$stationId,$time,&$updateTransaction){
                $updateTransaction[] = ["id"=>$obj->id,"fm_station_id"=>$stationId,"status"=>0,"updated_at"=>$time];
            },function ($service,$toLocation,$task,$prefix){
                return $service->fetchGroup_multiLocation($toLocation,$task,'','立架出至缓存架',20,false);
            },"fm_station_id")){
                Log::debug("队列事务5",["缓存架剩余数量：{$stations->count()}"]);
                DB::commit();return $stations->count()==0; //缓存架用完 跳出，否则接着分发
            }
            DB::rollBack();
        }catch (\Exception $e){
            DB::rollBack();
            $this->push(__METHOD__."->".__LINE__,"入库队列执行错误",$e->getMessage()." | 当前任务数：".$this->count." | 缓存信息：".(isset($available) ? json_encode($available) : ''));
        }
        return false;
    }

    private function dispatchTask(\Illuminate\Database\Eloquent\Collection $tasks,&$stations, $service,
                                  \Closure $update, \Closure $execute, string $stationName):bool
    {
        $locations = $stations;
        if ($tasks->count()>$locations->count())$tasks = $tasks->slice(0,$locations->count());//事务过多切割部分处理
        if ($tasks->count()<$locations->count()){
            $stations = $stations->slice($tasks->count());
            $stations = $stations->values($stations);
        }
        $toLocation = collect();
        $task = collect();
        $updateTask = [["id","station_id","updated_at"]];
        $updateTransaction = [["id",$stationName,"status","updated_at"]];
        $time = date("Y-m-d H:i:s");
        $map = [];
        foreach ($tasks as $index=>$obj){
            $toLocation->push($stations[$index]->code);
            $map[$stations[$index]->code] = $stations[$index]->id;
            $obj->task->station_id = $stations[$index]->id;
            $task->push($obj->task);
            $updateTask[] = ["id"=>$obj->task->id,"station_id"=>$stations[$index]->id,"updated_at"=>$time];
            $update($obj,$stations[$index]->id,$time,$updateTransaction);
        }
        app("BatchUpdateService")->batchUpdate("station_task_material_boxes",$updateTask);
        app("BatchUpdateService")->batchUpdate("task_transactions",$updateTransaction);
        if ($execute($service,$toLocation,$task,$tasks[0]->station_task_batch_id ?: '')){
            foreach ($toLocation as $value){
                app("CacheShelfService")->lightUp($value,'3','0',["title"=>"机器人取箱中，禁止操作"]);
                Cache::forever("CACHE_SHELF_OCCUPANCY_{$map[$value]}",true);
            }
            app("StationService")->locationOccupyMulti($toLocation->toArray());
            DB::commit();
            return true;
        }
        DB::rollBack();
        $this->push(__METHOD__."->".__LINE__,"缓存队列执行错误","库位信息：".json_encode($toLocation)."  任务信息：".json_encode($task));
        return false;
    }
}