simps/mqtt:适用于PHP的 MQTT 协议解析和协程客户端

简介: Simps 的第一个版本 MQTT 库 就是参考了 Workerman 的实现,使其能够使用 Swoole 的协程能力,同时也修复了一些问题

MQTT 是一种基于发布/订阅(publish/subscribe)模式的"轻量级"通讯协议,作为一种低开销、低带宽占用的即时通讯协议,已经成为物联网的重要组成部分


Swoole 也给 PHP 提供了开发物联网项目的能力,只需要设置一个 open_mqtt_protocol 选项,启用后就会解析 MQTT 包头,在 Worker 进程的 onReceive 事件每次都会返回一个完整的 MQTT 数据包


当然其他的也有,例如 Workerman 之前提供的 异步 mqtt 客户端库 ,还有其他的开源库,这里就不一一介绍了


Simps 的第一个版本 MQTT 库 就是参考了 Workerman 的实现,使其能够使用 Swoole 的协程能力,同时也修复了一些问题


第一个版本的实现是放在了框架当中,限制了一些用户的使用。于是又开始了重构,将 MQTT 独立为一个 library ,方便用户使用的同时也丰富了 PHP 生态,让 PHP 程序员不再局限于 Web 开发


在第一个版本发布之后,Simps 的交流群中也有不少用户询问 MQTT 的问题,Swoole 也修复了一些相关的 Bug,现在使用 PHP + Swoole 去开发物联网相关的项目应该是如虎添翼


同时第一个版本的 MQTT 库,只支持 MQTT 3.x,不支持 MQTT 5.0,在 GitHub 上也没有找到相关支持的类库,所以在重构了 3.x 版本之后,也支持了一下 MQTT 5.0


也许这是第一个支持 MQTT v5.0 协议的 PHP library...


支持 MQTT 协议 3.1、3.1.1 和 5.0 版本,支持 QoS 0、QoS 1、QoS 2,那么它来了,使用 composer 来安装


composer require simps/mqtt

安装成功之后我们来看一下订阅和发布的使用,以 MQTT5.0 为例


订阅


首先应该是订阅,订阅成功之后才能收到对应主题的发布消息,创建一个subscribe.php写入以下内容

include __DIR__ . '/vendor/autoload.php';
use Simps\MQTT\Hex\ReasonCode;
use Swoole\Coroutine;
use Simps\MQTT\Client;
use Simps\MQTT\Types;
$config = [
    'host' => 'broker.emqx.io',
    'port' => 1883,
    'time_out' => 5,
    'user_name' => 'user001',
    'password' => 'hLXQ9ubnZGzkzf',
    'client_id' => Client::genClientID(),
    'keep_alive' => 10,
    'properties' => [
        'session_expiry_interval' => 60,
        'receive_maximum' => 200,
        'topic_alias_maximum' => 200,
    ],
    'protocol_level' => 5,
];
Coroutine\run(function () use ($config) {
    $client = new Client($config, ['open_mqtt_protocol' => true, 'package_max_length' => 2 * 1024 * 1024]);
    while (!$data = $client->connect()) {
        Coroutine::sleep(3);
        $client->connect();
    }
    $topics['simps-mqtt/user001/get'] = [
        'qos' => 1,
        'no_local' => true,
        'retain_as_published' => true,
        'retain_handling' => 2,
    ];
    $timeSincePing = time();
    $res = $client->subscribe($topics);
    // 订阅的结果
    var_dump($res);
    while (true) {
        $buffer = $client->recv();
        if ($buffer && $buffer !== true) {
            $timeSincePing = time();
            // 收到的数据包
            var_dump($buffer);
        }
        if (isset($config['keep_alive']) && $timeSincePing < (time() - $config['keep_alive'])) {
            $buffer = $client->ping();
            if ($buffer) {
                echo 'send ping success' . PHP_EOL;
                $timeSincePing = time();
            } else {
                $client->close();
                break;
            }
        }
        // QoS1 发布回复
        if ($buffer['type'] === Types::PUBLISH && $buffer['qos'] === 1) {
            $client->send(
                [
                    'type' => Types::PUBACK,
                    'message_id' => $buffer['message_id'],
                    'code' => ReasonCode::SUCCESS
                ]
            );
        }
    }
});


执行php subscribe.php,就会得到这样的输出


array(3) {
  ["type"]=>
  int(9)
  ["message_id"]=>
  int(1)
  ["codes"]=>
  array(1) {
    [0]=>
    int(1)
  }
}

表示订阅成功,codes 对应的是对应订阅主题的 QoS 等级


发布


订阅成功之后,创建一个publish.php来测试发布

include __DIR__ . '/vendor/autoload.php';
use Swoole\Coroutine;
use Simps\MQTT\Client;
$config = [
    'host' => 'broker.emqx.io',
    'port' => 1883,
    'time_out' => 5,
    'user_name' => 'user002',
    'password' => 'adIJS1D482sd',
    'client_id' => Client::genClientID(),
    'keep_alive' => 20,
    'properties' => [
        'session_expiry_interval' => 60,
        'receive_maximum' => 200,
        'topic_alias_maximum' => 200,
    ],
    'protocol_level' => 5,
];
Coroutine\run(function () use ($config) {
    $client = new Client($config, ['open_mqtt_protocol' => true, 'package_max_length' => 2 * 1024 * 1024]);
    while (!$client->connect()) {
        Coroutine::sleep(3);
        $client->connect();
    }
    while (true) {
        $response = $client->publish(
            'simps-mqtt/user001/get',
            '{"time":' . time() . '}',
            1,
            0,
            0,
            ['topic_alias' => 1]
        );
        var_dump($response);
        Coroutine::sleep(3);
    }
});


代码的意思是每隔 3 秒给订阅的主题simps-mqtt/user001/get发布一次消息

打开一个新的终端窗口,执行php publish.php就会得到输出:


array(4) {
  ["type"]=>
  int(4)
  ["message_id"]=>
  int(1)
  ["code"]=>
  int(0)
  ["message"]=>
  string(7) "Success"
}


这里增加了 message,为了用户可读,不需要去查找对应的 code 含义是什么

返回到订阅的窗口,就会看到所打印的发布信息


array(8) {
  ["type"]=>
  int(3)
  ["topic"]=>
  string(0) ""
  ["message"]=>
  string(19) "{"time":1608017156}"
  ["dup"]=>
  int(1)
  ["qos"]=>
  int(1)
  ["retain"]=>
  int(0)
  ["message_id"]=>
  int(4)
  ["properties"]=>
  array(1) {
    ["topic_alias"]=>
    int(1)
  }
}

这样一个简单的发布订阅功能就实现了

在这个库中还有一些值得优化和还未完成的部分,如还没有支持 MQTT5 的Auth type,以及部分的properties还未支持

想参与的同学可以提交 PR,如果有问题也可以提交 Issue,让我们共同去建设 PHP 的生态

相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
3月前
|
JSON 定位技术 PHP
PHP技巧:解析JSON及提取数据
这就是在PHP世界里探索JSON数据的艺术。这场狩猎不仅仅是为了获得数据,而是一种透彻理解数据结构的行动,让数据在你的编码海洋中畅游。通过这次冒险,你已经掌握了打开数据宝箱的钥匙。紧握它,让你在编程世界中随心所欲地航行。
140 67
|
2月前
|
网络协议 Shell PHP
简单的php版本nacos客户端
这是一个简单的 PHP 版本 Nacos 客户端,支持服务注册、配置发布、服务发现等功能。通过 Composer 安装,提供服务端与客户端示例代码,可快速集成至项目中。适用于基于 Nacos 的微服务架构开发,帮助实现服务治理与配置管理。
59 5
|
6月前
|
消息中间件 存储 数据采集
4步实现状态机驱动的MQTT客户端,快速接入OneNet (1)
本文介绍了基于状态机驱动的MQTT客户端快速接入OneNet平台的实现方法,通过4步完成模块设计。文章以开源项目`Sparrow`为基础,引入`OneNetMqtt`业务模块,采用事件驱动模型和双层状态机设计,实现设备状态管理、消息处理及定时任务等功能。模块分为三层:`OneNetManager`负责核心逻辑,`OneNetDevice`管理设备信息,`OneNetDriver`处理Socket与MQTT通信。验证结果显示设备连接、数据上报及下线功能正常,稳定性良好。该设计简化了复杂条件判断,增强了系统灵活性与可扩展性,适用于实际项目参考。文末提供源码获取方式,助力读者实践与学习。
324 99
|
3月前
|
运维 监控 算法
局域网屏幕监控软件 PHP 图像块增量传输算法解析
本文探讨了一种基于PHP语言开发的图像块增量传输算法,适用于局域网屏幕监控场景。通过将屏幕图像分块处理、计算哈希值并对比变化区域,该算法显著降低了网络带宽占用,提升了监控效率。在企业管理和远程教育中,该技术可实现终端设备的实时监控与远程管控,同时支持与生物识别等技术融合,拓展应用范围。实验表明,该算法在常规办公场景下可减少90%以上的数据传输量,展现了良好的实时性和优化效果。
48 3
|
4月前
|
存储 监控 算法
内网监控桌面与 PHP 哈希算法:从数据追踪到行为审计的技术解析
本文探讨了内网监控桌面系统的技术需求与数据结构选型,重点分析了哈希算法在企业内网安全管理中的应用。通过PHP语言实现的SHA-256算法,可有效支持软件准入控制、数据传输审计及操作日志存证等功能。文章还介绍了性能优化策略(如分块哈希计算和并行处理)与安全增强措施(如盐值强化和动态更新),并展望了哈希算法在图像处理、网络流量分析等领域的扩展应用。最终强调了构建完整内网安全闭环的重要性,为企业数字资产保护提供技术支撑。
119 2
|
7月前
|
数据可视化 关系型数据库 MySQL
嵌入式C++、STM32、MySQL、GPS、InfluxDB和MQTT协议数据可视化
通过本文的介绍,我们详细讲解了如何结合嵌入式C++、STM32、MySQL、GPS、InfluxDB和MQTT协议,实现数据的采集、传输、存储和可视化。这种架构在物联网项目中非常常见,可以有效地处理和展示实时数据。希望本文能帮助您更好地理解和应用这些技术,构建高效、可靠的数据处理和可视化系统。
345 82
|
6月前
|
算法 测试技术 C语言
深入理解HTTP/2:nghttp2库源码解析及客户端实现示例
通过解析nghttp2库的源码和实现一个简单的HTTP/2客户端示例,本文详细介绍了HTTP/2的关键特性和nghttp2的核心实现。了解这些内容可以帮助开发者更好地理解HTTP/2协议,提高Web应用的性能和用户体验。对于实际开发中的应用,可以根据需要进一步优化和扩展代码,以满足具体需求。
523 29
|
6月前
|
存储 监控 算法
关于员工上网监控系统中 PHP 关联数组算法的学术解析
在当代企业管理中,员工上网监控系统是维护信息安全和提升工作效率的关键工具。PHP 中的关联数组凭借其灵活的键值对存储方式,在记录员工网络活动、管理访问规则及分析上网行为等方面发挥重要作用。通过关联数组,系统能高效记录每位员工的上网历史,设定网站访问权限,并统计不同类型的网站访问频率,帮助企业洞察员工上网模式,发现潜在问题并采取相应管理措施,从而保障信息安全和提高工作效率。
78 7
|
7月前
|
监控 Linux PHP
【02】客户端服务端C语言-go语言-web端PHP语言整合内容发布-优雅草网络设备监控系统-2月12日优雅草简化Centos stream8安装zabbix7教程-本搭建教程非docker搭建教程-优雅草solution
【02】客户端服务端C语言-go语言-web端PHP语言整合内容发布-优雅草网络设备监控系统-2月12日优雅草简化Centos stream8安装zabbix7教程-本搭建教程非docker搭建教程-优雅草solution
174 20
|
7月前
|
监控 关系型数据库 MySQL
【01】客户端服务端C语言-go语言-web端PHP语言整合内容发布-优雅草网络设备监控系统-硬件设备实时监控系统运营版发布-本产品基于企业级开源项目Zabbix深度二开-分步骤实现预计10篇合集-自营版
【01】客户端服务端C语言-go语言-web端PHP语言整合内容发布-优雅草网络设备监控系统-硬件设备实时监控系统运营版发布-本产品基于企业级开源项目Zabbix深度二开-分步骤实现预计10篇合集-自营版
130 0

推荐镜像

更多
  • DNS