workHour/app/Console/Commands/MqttSubscriber.php

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;
class MqttSubscriber extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'mqtt:subscribe';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = 'Command description';

    /**
     * Create a new command instance.
     *
     * @return void
     */
    public function __construct()
    {
        parent::__construct();
    }

    /**
     * Execute the console command.
     *
     * @return mixed
     */
    public function handle()
    {
        $server   = '47.111.77.194';
        $port     = 1883;
        $clientId = 'laravel_backend_worker';

        $mqtt = new MqttClient($server, $port, $clientId);

        $settings = (new ConnectionSettings)
            ->setUsername('php_server_user')
            ->setPassword('sEcrEt_pAss_123');

        $this->info("正在连接到 MQTT Broker...");

        $mqtt->connect($settings, true);

        // 订阅主题
        $mqtt->subscribe('/wy/119/RealtimeData/DT5/yonglidev1', function ($topic, $message) {
            $this->info("收到消息: $message");

            $data = json_decode($message, true);

            if (isset($data['total_work'])) {
                // 利用 Laravel 的模型直接入库
                $hours = round($data['total_work'] / 60, 2);

//                WorkLog::create([
//                    'device_id' => $data['device_id'] ?? 'unknown',
//                    'minutes'   => $data['total_work'],
//                    'hours'     => $hours,
//                    'raw_data'  => $message,
//                ]);

                $this->info("数据已入库: {$hours} 小时");
            }
        }, 0);

        // 开始死循环监听
        $mqtt->loop(true);
    }
}