PHP MQTT client客户端连接订阅/发布消息

我们先安装php-mqtt，命令如下：

composer require php-mqtt/client

封装代码如下：

use \PhpMqtt\Client\MqttClient;
use \PhpMqtt\Client\ConnectionSettings;

class Mqtt
{
    private static $instance;
    public $url;
    private $conn;

    private function __construct($config = [])
    {
        if (count($config) == 0) {
            $cfg = [
                'server'   => '127.0.0.1', //连接服务器地址
                'port'     => 1883, //连接端口
                'clientId' => time() . rand(5, 15), //客户端ID，可随意填写，也可使用rand函数生成随机的
                'username' => 'hilo8', //登录账号
                'password' => 'www.hilo8.com', //登录密码
                'clean_session' => false
            ];
        } else {
            $cfg = $config;
        }
        $connectionSettings = (new ConnectionSettings())
            ->setUsername($cfg['username'])
            ->setPassword($cfg['password'])
            ->setKeepAliveInterval(60);
        // Last Will 设置
        //->setLastWillTopic('topic/hsy/last-will')
        //->setLastWillMessage('client disconnect')
        //->setLastWillQualityOfService(1);
        $mqtt = new MqttClient($cfg['server'], $cfg['port'], $cfg['clientId']);
        $mqtt->connect($connectionSettings, $cfg['clean_session']);
        $this->conn = $mqtt;
    }

    //连接MQTT
    public static function connection($config = [])
    {
        if (!self::$instance instanceof self) {
            self::$instance = new self($config);
        }
        return self::$instance;
    }

    //订阅主题，参数$topic=主题名称，$url=收到信息POST发送到的链接地址
    public function subscribe($topic, $url = '')
    {
        $this->url = $url;
        $this->conn->subscribe($topic, function ($topic, $message) {
            $data = [
                'topic' => $topic,
                'message' => $message
            ];
            if ($this->url != '') {
                $curl = curl_init();
                curl_setopt_array($curl, array(
                    CURLOPT_URL => $this->url,
                    CURLOPT_RETURNTRANSFER => true,
                    CURLOPT_ENCODING => "",
                    CURLOPT_MAXREDIRS => 10,
                    CURLOPT_TIMEOUT => 0,
                    CURLOPT_FOLLOWLOCATION => true,
                    CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
                    CURLOPT_CUSTOMREQUEST => "POST",
                    CURLOPT_POSTFIELDS => $data
                ));
                $response = curl_exec($curl);
                curl_close($curl);
                //echo $response;
            } else {
                echo json_encode($data, JSON_UNESCAPED_UNICODE);
            }
        }, 0);
        $this->conn->loop(true);
    }

    //发布消息，参数$topic=主题名称，$message=要发布的消息(格式string/number/array)
    public function publish($topic, $message)
    {
        $payload = is_array($message) ? json_encode($message, JSON_UNESCAPED_UNICODE) : $message;
        $this->conn->publish($topic, $payload, 0, true);
        $this->conn->disconnect(); //断开 MQTT 连接
        return true;
    }
}

订阅：

参考来源：https://www.emqx.com/zh/blog/how-to-use-mqtt-in-php