script_backend/application/jobs/DeviceReportEvent.php

<?php

namespace app\jobs;

use app\admin\model\Device;
use think\Exception;
use think\queue\Job;

class DeviceReportEvent
{

    public function fire(Job $job, $data)
    {
        // 拆解是哪个设备发送的消息
        $topic = $data['topic'];

        $topicArray = explode('/', $topic);
        $device_id = $topicArray[2];
        $reportData = json_decode($data['message'], true);
        try {
            $this->eventRouter($device_id, $reportData);
        } catch (Exception $exception) {
            // 发送错误，跳过， 这里可以记录一下日志
        }
        $job->delete();
    }

    protected function eventRouter($device_id, $data)
    {
        // 消息体结构应该如下：
//        [
//            "TYPE"  => "ONLINE", // 事件类型
//            "DATA"  =>  [] // 事件参数
//        ];

        if (!isset($data['TYPE'])) return false;

        switch ($data['TYPE']) {
            case "ONLINE":
                $this->online($device_id, $data['DATA']);
                break;
        }
        return true;
    }

    protected function online($device_id, $data)
    {
        $device = Device::where("device_sn", $device_id)->find();
        if ($device) {
            $device->status = 1;
            $device->save();
        } else {
            Device::insertGetId([
                "device_sn" => $device_id,
                "status" => 1
            ]);
        }
    }
}