script_backend/application/admin/services/MqttMessageClient.php

<?php

namespace app\admin\services;

use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\Exceptions\ProtocolNotSupportedException;
use think\Env;

class MqttMessageClient
{
    protected static $instance = null;

    protected $client = null;

    /**
     * 功能：实例化mqtt
     * @throws ProtocolNotSupportedException
     * @throws \PhpMqtt\Client\Exceptions\ConfigurationInvalidException
     * @throws \PhpMqtt\Client\Exceptions\ConnectingToBrokerFailedException
     */
    protected function __construct()
    {
//        $protocol = Env::get("mqtt.protocol");
        $clientId = Env::get("mqtt.client_id");
        $hostname = Env::get("mqtt.hostname");
        $port = Env::get("mqtt.port");
        $username = Env::get("mqtt.username");
        $password = Env::get("mqtt.password");


        $mqtt = new MqttClient($hostname, $port, $clientId, MqttClient::MQTT_3_1_1);
        $connectionSettings = (new ConnectionSettings());
        if ($username !== null) {
            $connectionSettings = $connectionSettings->setUsername($username);
        }
        if ($password !== null) {
            $connectionSettings = $connectionSettings->setPassword($password);
        }

        $this->client = $mqtt;
        $mqtt->connect($connectionSettings);
    }

    protected function __clone()
    {

    }

    public static function getInstance()
    {
        if (static::$instance == null) {
            static::$instance = new static();
        }
        return static::$instance;
    }

    public function subscribe($topic, $callback)
    {
        $this->client->subscribe($topic, $callback);
    }

    public function publish(string $topic, string $message, int $qos = 0, bool $retain = false)
    {
        $this->client->publish($topic, $message, $qos, $retain);
    }

    public function loop(bool $keepAlive = true): void
    {
        $this->client->loop($keepAlive);
    }
}