MQTT 是一种基于发布/订阅模式的 轻量级物联网消息传输协议 ，可以用极少的代码和带宽为联网设备提供实时可靠的消息服务，它广泛应用于物联网、移动互联网、智能硬件、车联网、电力能源等行业。

本文主要介绍如何在 PHP项目中使用composer require php-mqtt/client库 ，实现客户端与 MQTT 服务器 的连接、订阅、收发消息等功能。

<?php

namespace app\command;

use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\MqttClient;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use util\LogHelperUtil;

// composer require php-mqtt/client
class Mqtt extends Command
{
    protected function configure()
    {
        $this->setName('mqtt')
            ->setDescription('mqtt Hello');
    }

    protected function execute(Input $input, Output $output)
    {
        $mqttConfig = config('mqtt');
        // MQTT代理的配置
        $server = $mqttConfig['host'] ?? ''; // MQTT代理的地址
        $port = $mqttConfig['port'] ?? 1883; // MQTT代理的端口
        $username = 'test_mqtt'; // MQTT代理的用户名（如果需要）
        $password = '123456'; // MQTT代理的密码（如果需要）
        $clientId = 'service-mqtt-' . time(); // 客户端ID
        try {
            // 创建MQTT客户端实例
            $mqtt = new MqttClient($server, intval($port), $clientId);
            $settings = (new ConnectionSettings())->setUsername($username)->setPassword($password)
                ->setKeepAliveInterval(10)   // 根据需要设置心跳间隔
                ->setReconnectAutomatically(true) // 是否会尝试自动重新连接
                ->setDelayBetweenReconnectAttempts(2000) // 定义重新连接尝试之间的延迟（毫秒）。
                ->setMaxReconnectAttempts(10); // 重新连接的最大尝试次数
            $mqtt->connect($settings); // 连接到MQTT代理

            // 订阅一个主题
            $topic = 'testtopic/#';
            $mqtt->subscribe($topic, function ($topic, $message) use ($mqtt) {
                $time = self::getTime("Y年m月d日G时i分s秒x毫秒");
                echo "{$time} 主题：{$topic} - 收到：" . ($message) . PHP_EOL;
            
                if ($topic != 'testtopic/service/2' && $topic != 'testtopic/golang') {
                    $mqtt->publish('testtopic/service/2', "hello[{$topic}]", 2, true);
                }
            }, 2);

            // 保持脚本运行，以便接收消息
//            while ($mqtt->isConnected()) {
//                $mqtt->loop();
//            }
            $mqtt->loop();
//            $mqtt->disconnect(); // 断开连接
        } catch (\Exception $e) {
            echo "MQTT Error: " . $e->getMessage();
        }

        $output->writeln("Received message on topic");
    }

    private static function getTime($tag)
    {
        list($usec, $sec) = explode(" ", microtime());
        $now_time = $sec . '.' . substr($usec, 2, 4);
        list($usec, $sec) = explode(".", $now_time);
        $date = date($tag, $usec);
        return str_replace('x', $sec, $date);
    }
}