代码收藏家技术教程 2025-02-26

MQTT开发，thinkphp6配合workerman/mqtt做物联网mqtt订阅与发布消息，支持长连接、断开重连

提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

文章目录

前言

一、部署thinkphp6

二、配置文件

1.用[Visual Studio Code](https://code.visualstudio.com/)打开tp项目文件

2.编写Mqtt类

3.运行并测试监听主题消息是否正常

3.1 在tp根目录执行启动命令

3.2 在Broker服务端看是否链接成功

4.编写测试方法测试mqtt发布消息

总结

前言

提示：这篇文章主在帮助使用tp6开发mqtt物联网网友：

首先：你要明白什么是mqtt,建议先看官网文档：MQTT 协议快速入门：基础知识和实用教程 | EMQ
其次：要会thinkphp6:ThinkPHP官方手册
最后你要了解workerman/mqtt是怎么简单使用的:workerman/mqtt

提示：开始搭建部署

一、部署thinkphp6

使用composer安装tp6：不了解composer的请先学习:Composer 中文网致力于推广 PHP 世界的包管理工具

# 安装命令
composer create-project topthink/think=6.0.x tp

完成之后进入tp目录

然后安装 topthink/think-worker

# 安装命令
composer require topthink/think-worker

其次再安装workerman/mqtt

# 安装命令
composer require workerman/mqtt

二、配置文件

1.用Visual Studio Code打开tp项目文件

打开config下的worker_server.php的配置文件

配置项worker_class

 'worker_class'   => [ // 自定义Workerman服务类名 支持数组定义多个服务
        'mqtt\Mqtt'
    ],

如图

2.编写Mqtt类

在extend目录创建mqtt目录，在mqtt下创建Mqtt类

编写Mqtt类

namespace mqtt;

use think\facade\Log;
use think\worker\Server;
use Workerman\Worker;

// mqtt类继承think\worker\Server
class Mqtt extends Server
{
    private $connection = null;
    public function onWorkerStart($worker)
    {
        $options = [
            'keepalive'=>60,
            'client_id'=>"PUBLISH002",
            'clean_session'=>true,
            'reconnect_period'=>10,
            'username'=>"admin",
            "password"=>"Admin@1234567890"
        ];
        $mqtt = new \Workerman\Mqtt\Client('mqtt://mqt.test.com:1883',$options); // mqtt://mqt.test.com:1883 这个域名请填写自己Broker的 域名
        $this->connection = $mqtt;
        // 订阅mqtt主题消息
        $mqtt->onConnect = function($mqtt) {
            $mqtt->subscribe('pub/#');// 订阅mqtt主题 pub/# 根据实际开发修改
        };
        $mqtt->onMessage = function($topic, $content){// 在此监听主题消息
            $data = json_decode($content,true);
            print_r('------订阅的主题消息-------');
            var_dump($data);
            //这里是你的业务代码下面是我的业务代码，根据时间需求开发即可
            Log::write('onWorkerStart==:'.var_export($data,true));// 打印日志看看
        };
        $mqtt->connect();

        
        // 内部端口调用 用于本地发布消息的时候调用
        $inner_text_worker = new Worker('text://0.0.0.0:1884'); //内部调用端口
        $inner_text_worker->onMessage = function ($connection, $message){
            print_r('------发布消息-------');
            var_dump($message);
            $this->connection->publish('sub/PUBLISH001', $message);
            $connection->send('ok');
        };
        $inner_text_worker->listen();
    }
}

3.运行并测试监听主题消息是否正常

首先需要有自己的MQTT Broker服务，没有的不会的请先了解https://www.emqx.com/zh/blog/the-ultimate-guide-to-mqtt-broker-comparison,或者搭建自己的私有MQTT-broke服务
至此Mqtt类已经写好

3.1 在tp根目录执行启动命令

# 启动命令 如果要守护的话 在命令后面加 -d 如：php think worker:server -d
php think worker:server

如果启动出现 Error: Class ‘Workerman\Timer’ not found in F:\study\mqtt\tp\vendor\workerman\mqtt\src\Client.php:877 如图

那么说明安装Workerman\mqtt的版本出现点问题
修改一下文件workerman\mqtt\src\Client.php

use \Workerman\Protocols\Mqtt;
use \Workerman\Timer;
// 修改后
use \Workerman\Mqtt\Protocols\Mqtt;
use \Workerman\Lib\Timer;

如图

然后再次执行启动命令php think worker:server

这样就启动成功了

3.2 在Broker服务端看是否链接成功

不错，已经了成功，测试一下订阅的主题是否能收到消息

主题订阅的消息也没有问题，链接也很稳定

4.编写测试方法测试mqtt发布消息

在app/controller/Index.php文件中编写一个测试方法sendMsg 发布消息

 public function sendMsg()
    {
        $arr = ['test'=>'发布消息测试！'];
        if(empty($arr)) return false;
        try {
            $client = stream_socket_client('tcp://127.0.0.1:1884', $errno, $errmsg, 1);
        }catch (\ErrorException $e){
            return false;
        }
        fwrite($client, json_encode($arr) . "\n");
        $_ok = trim(fread($client, 8192));//8192读取数据字节的长度
        dd($_ok);
    }

如图

终端运行 php think run 启动项目访问 http://127.0.0.1:8000/index/sendMsg

看上图发布消息也是可以的，至此tp6配合workerman/mqtt做物联网mqtt订阅与发布消息简单开发就介绍到这里