data = $data; $this->url = config('ip.zslf'); } /** * Execute the job. * * @return void */ public function handle() { try { if (empty($this->data['data'])) return; $deviceId = $this->data['data']['deviceId']; $dataPoints = $this->data['data']['dataPoints'] ?? []; if ($this->data['type'] == "dataPoint") { $redisKey = 'buffer:jialiwumei_data'; $batchSize = 500; // 1. 整理数据并推入 Redis 列表 $records = []; foreach ($dataPoints as $value) { if (empty($value['value'])) continue; $records[] = json_encode([ 'machine_code' => $deviceId . $value['dataPointId'], 'param_value' => floatval($value['value']), 'time' => $this->getNowDay() ]); } if (!empty($records)) { // 批量推入 Redis \Illuminate\Support\Facades\Redis::rpush($redisKey, ...$records); // 【核心修改点】:每次写入数据都重置 Key 的过期时间为 30 分钟 (1800秒) // 只要持续有数据进来,这个 Key 就一直有效 // 如果超过 30 分钟没新数据进来(设备关机/下班),Redis 自动删除该 Key \Illuminate\Support\Facades\Redis::expire($redisKey, 1800); } // 2. 检查 Redis 里的总条数 $currentCount = \Illuminate\Support\Facades\Redis::llen($redisKey); // 3. 达到 500 条,执行批量发送 if ($currentCount >= $batchSize) { $this->processBatchSend($redisKey, $batchSize); } } } catch (\Exception $exception) { Log::channel('apiLog')->info('jlwm开始位置', ["param" => $exception->getMessage()]); $this->delete(); } } /** * 批量处理发送逻辑 */ protected function processBatchSend($redisKey, $batchSize) { // 原子化弹出 500 条数据 $rawRecords = \Illuminate\Support\Facades\Redis::pipeline(function ($pipe) use ($redisKey, $batchSize) { for ($i = 0; $i < $batchSize; $i++) { $pipe->lpop($redisKey); } }); $batchData = array_map(fn($item) => json_decode($item, true), array_filter($rawRecords)); if (empty($batchData)) return; list($status,$msg) = $this->sendToDeviceBatch($batchData); if(! $status) echo $msg; } public function sendToDeviceBatch($data){ list($status,$token) = ClearDataService::getTokenJLWM(); if(! $status) return [false, $token]; $url = $this->url . "api/module-data/jlwm_device_machine_record/jlwm_device_machine_record/diy/create_data"; $post = [ 'data' => [ 'device_machine_record' => $data ] ]; $header = ["Authorization: Bearer {$token}","Content-Type:application/json","Site:JLWM"]; $curl = curl_init(); curl_setopt_array($curl, array( CURLOPT_URL => $url, CURLOPT_RETURNTRANSFER => true, CURLOPT_ENCODING => '', CURLOPT_MAXREDIRS => 10, CURLOPT_TIMEOUT => 15, CURLOPT_FOLLOWLOCATION => true, CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1, CURLOPT_CUSTOMREQUEST => 'POST', CURLOPT_POSTFIELDS => json_encode($post), CURLOPT_SSL_VERIFYPEER => false, CURLOPT_HTTPHEADER => $header, )); $response = curl_exec($curl); if ($response === false) { // 获取错误号 $errorNumber = curl_errno($curl); // 获取错误信息 $errorMessage = curl_error($curl); $message = "cURL Error #{$errorNumber}: {$errorMessage}"; Log::channel('apiLog')->info('jlwm写入数据', ["param" => $message]); } curl_close($curl); $res = json_decode($response,true); if(! isset($res['status'])){//报错了 Log::channel('apiLog')->info('jlwm写入数据返回错误', ["param" => $response]); } return [true, '']; } protected function echoMessage(OutputInterface $output) { $output->writeln($this->data); } function getNowDay(){ // 获取当前时间 $currentDateTime = new \DateTime(); // 设置时区为 PRC $currentDateTime->setTimezone(new \DateTimeZone('UTC')); // 格式化时间为 "2023-09-21T16:00:00.000Z" 的格式 $formattedDateTime = $currentDateTime->format('Y-m-d\TH:i:s.000\Z'); return $formattedDateTime; } }