script_backend/application/admin/command/MqttMessage.php

<?php

namespace app\admin\command;

use app\admin\services\MqttMessageClient;
use app\jobs\DeviceReportEvent;
use \PhpMqtt\Client\MqttClient;
use \PhpMqtt\Client\ConnectionSettings;
use think\console\Command;
use think\console\Input;
use think\console\input\Option;
use think\console\Output;
use think\Env;
use think\Queue;

class MqttMessage extends Command
{
    protected $client = null;

    public function __construct($name=null)
    {
        parent::__construct($name);
        $this->client = MqttMessageClient::getInstance();
    }

    public function onMessage($topic, $message) {
        var_dump($topic);
        $jobHandlerClassName = DeviceReportEvent::class;
        $jobQueueName  = "DeviceEventQueue";
        $jobData = [
            "topic" =>  $topic,
            "message"   =>  $message
        ];
        $isPushed = Queue::push( $jobHandlerClassName , $jobData , $jobQueueName );
        if( $isPushed !== false ){
            echo '消息已发出';
        }else{
            echo '消息发送出错';
        }
    }
    public function configure()
    {
        $this
            ->setName('MqttMessage')
            ->setDescription('读取MQTT消息');
    }

    public function execute(Input $input, Output $output)
    {

        $this->client->subscribe("/device/+/report", function ($topic, $message){
            $this->onMessage($topic, $message);
        });
        $this->client->loop(true);
    }
}