thinkphp6安装php-mqtt/client,并实现实时消息收发写入日志

thinkphp6安装php-mqtt/client,并实现实时消息收发写入日志

系统：centos7

第一步：宝塔面板安装php环境8.0；

第二步：宝塔自带安装composer;

第三步：下载thinkphp6

create project composer require topthink/think:6.0.0 tp

tp是 新建的站点名称，设置php的访问目录到public下的index.php的入口文件；
建议开启：调试模式；

第四步 下载

composer require php-mqtt/client

参考地址和用法：https://gitcode.com/gh_mirrors/client9/client/blob/master/tests/TestCase.php

第五步：重点来了：
我这里采用单应用的模式（多应用的自行调整）：
5.0文档编辑完了，发现发送主体没有编辑，来把 🐆补个；

<?php
namespace app\controller;

use app\BaseController;
use Exception;
use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\Exceptions\ConfigurationInvalidException;
use PhpMqtt\Client\Exceptions\ConnectingToBrokerFailedException;
use PhpMqtt\Client\Exceptions\DataTransferException;
use PhpMqtt\Client\Exceptions\ProtocolNotSupportedException;
use PhpMqtt\Client\Exceptions\RepositoryException;
use React\EventLoop\Factory;
use think\facade\Log;

class Index extends BaseController
{
    private $server = '118.*******';
    private $port = 1883;
    private $clientId = 'adminserver';
    private $username = 'adminserver';
    private $password = 'adminserver';

    /**
     * @throws ConnectingToBrokerFailedException
     * @throws ConfigurationInvalidException
     * @throws RepositoryException
     * @throws ProtocolNotSupportedException
     * @throws DataTransferException
     */
    public function index()
    {
//        测试消息发送ok
        $this->publish('testtopic/1', 'Hello, MQTT!');
    }
public function publish($topic, $message, $qos = 2)
    {
        try {
            // 创建 MQTT 客户端实例
            $mqtt = new MqttClient($this->server, $this->port, $this->clientId);

            // 设置连接选项
            $connectionSettings = (new ConnectionSettings)
                ->setUsername($this->username) // 如果需要认证，设置用户名
                ->setPassword($this->password) // 如果需要认证，设置密码
                ->setKeepAliveInterval(60); // 设置心跳间隔
//                ->setReconnectAutomatically(true); // 设置为干净会话

            // 连接到 MQTT 服务器
            $mqtt->connect($connectionSettings, true);

            // 发布消息
            $mqtt->publish($topic, $message, $qos);

            // 断开连接
            $mqtt->disconnect();

            echo "消息已成功发布到主题: $topic\n";
        } catch (\Exception $e) {
            // 捕获并处理异常
            echo "发生错误: " . $e->getMessage() . "\n";
        }
    }

5.1/yourprject/app/command/定义：MqttSubscribe.php

<?php
namespace app\command;

use think\console\Command;
use think\console\Input;
use think\console\Output;
use PhpMqtt\Client\MqttClient;
use PhpMqtt\Client\ConnectionSettings;
use PhpMqtt\Client\Exceptions\ConfigurationInvalidException;
use PhpMqtt\Client\Exceptions\ConnectingToBrokerFailedException;
use PhpMqtt\Client\Exceptions\DataTransferException;
use PhpMqtt\Client\Exceptions\ProtocolNotSupportedException;
use PhpMqtt\Client\Exceptions\RepositoryException;
use think\facade\Log;

class MqttSubscribe extends Command
{
    protected function configure()
    {
        $this->setName('mqtt:subscribe')
             ->setDescription('Subscribe to an MQTT topic');
    }

    protected function execute(Input $input, Output $output)
    {
        $server   = '118.***'; // MQTT 服务器地址
        $port     = 1883;        // MQTT 服务器端口
        $clientId = 'adminserver'; // 客户端 ID

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

        $connectionSettings = (new ConnectionSettings)
            ->setUsername($clientId) // 可选
            ->setPassword($clientId) // 可选
            ->setKeepAliveInterval(60);             // 可选
            // ->setConnectTimeout(10)        // 可选
            // ->setWriteTimeout(5);          // 可选

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

        $topic = 'test/topic';

        // $mqtt->subscribe($topic, function ($topic, $message) {
        //     log::info("Received message on topic [$topic]: $message");
        //     // 进行数据处理操作
        // }, 0);
        
        $mqtt->subscribe($topic, function ($topic, $message) use ($output) {
            echo "Received message on topic [{$topic}]: {$message}";
            // 写入日志
            log::info("Received message on topic [{$topic}]: {$message}");

            // 控制台输出
            

            // 进行数据处理操作
            // 例如：处理接收到的消息
        }, 0);


        $mqtt->loop(true);

        $mqtt->disconnect();
    }
}

5.2/yourprject/app/command/定义：Command.php

<?php
namespace app\command;

use think\console\Command as BaseCommand;
use think\console\Input;
use think\console\Output;
use think\facade\App;

class Command extends BaseCommand
{
    protected function configure()
    {
        $this->addCommands([
            'app\command\MqttSubscribe',
        ]);
    }
}

5.3/yourprject/config/修改，如果没有就创建：Command.php

<?php

return [
    'commands' => [
        'mqtt:subscribe' => 'app\command\MqttSubscribe',
    ],
];

5.4/yourprject/config/console.php：

<?php
// +----------------------------------------------------------------------
// | 控制台配置
// +----------------------------------------------------------------------
return [
    'commands' => [
        \app\command\MqttSubscribe::class,
    ],
];

```powershell
提示：
//重新生成命令：
php think optimize

//启动 
php think mqtt:subscribe

//清除会话
php think clear

后端实时接收效果截图

客户端发起主体截图

日志与终端同时写入的效果图

6.至此，实时接收做完成（注意这个是持续会话连接，无需再次启停），接下来是php封装调用；

7.封装报错：
Call to undefined function app\controller\shell_exec()
7.1解决方案：用官方文档的办法,现在需要解决守护进程和动态传入主题；