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

本文涉及的产品
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
全局流量管理 GTM,标准版 1个月
云解析 DNS,旗舰版 1个月
简介: 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版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
21天前
|
XML 监控 网络协议
云深处绝影四足机器人协议学习解析
本文详细介绍并解析了云深处绝影X20四足机器人的通信协议,包括TCP服务端端口号、基于Service的请求/响应通信机制、通信帧结构、消息类型、常见的通信示例如获取状态和导航请求,以及运动控制的参数和命令。文中还提出了对协议中某些未明确说明或可能存在的问题的疑惑。
25 0
云深处绝影四足机器人协议学习解析
|
17天前
|
缓存 网络协议 安全
【网络攻防战】DNS协议的致命弱点:如何利用它们发动悄无声息的网络攻击?
【8月更文挑战第26天】DNS(域名系统)是互联网的关键组件,用于将域名转换为IP地址。然而,DNS协议存在安全漏洞,包括缺乏身份验证机制、缓存中毒风险及放大攻击的可能性。通过具体案例,如DNS缓存中毒和DNS放大攻击,攻击者能够误导用户访问恶意站点或对目标服务器实施DDoS攻击。为了防范这些威胁,可以采用DNSSEC实现数字签名验证、利用加密的DNS服务(如DoH或DoT)、限制DNS服务器响应以及及时更新DNS软件等措施。理解并应对DNS的安全挑战对于确保网络环境的安全至关重要。
49 2
|
19天前
|
消息中间件 开发者
【RabbitMQ深度解析】Topic交换器与模式匹配:掌握消息路由的艺术!
【8月更文挑战第24天】在消息队列(MQ)体系中,交换器作为核心组件之一负责消息路由。特别是`topic`类型的交换器,它通过模式匹配实现消息的精准分发,适用于发布-订阅模式。不同于直接交换器和扇形交换器,`topic`交换器支持更复杂的路由策略,通过带有通配符(如 * 和 #)的模式字符串来定义队列与交换器间的绑定关系。
36 2
|
11天前
|
消息中间件 Kafka Java
Spring 框架与 Kafka 联姻,竟引发软件世界的革命风暴!事件驱动架构震撼登场!
【8月更文挑战第31天】《Spring 框架与 Kafka 集成:实现事件驱动架构》介绍如何利用 Spring 框架的强大功能与 Kafka 分布式流平台结合,构建灵活且可扩展的事件驱动系统。通过添加 Spring Kafka 依赖并配置 Kafka 连接信息,可以轻松实现消息的生产和消费。文中详细展示了如何设置 `KafkaTemplate`、`ProducerFactory` 和 `ConsumerFactory`,并通过示例代码说明了生产者发送消息及消费者接收消息的具体实现。这一组合为构建高效可靠的分布式应用程序提供了有力支持。
38 0
|
14天前
|
消息中间件 Java RocketMQ
微服务架构师的福音:深度解析Spring Cloud RocketMQ,打造高可靠消息驱动系统的不二之选!
【8月更文挑战第29天】Spring Cloud RocketMQ结合了Spring Cloud生态与RocketMQ消息中间件的优势,简化了RocketMQ在微服务中的集成,使开发者能更专注业务逻辑。通过配置依赖和连接信息,可轻松搭建消息生产和消费流程,支持消息过滤、转换及分布式事务等功能,确保微服务间解耦的同时,提升了系统的稳定性和效率。掌握其应用,有助于构建复杂分布式系统。
31 0
|
18天前
|
安全 5G 数据安全/隐私保护
|
18天前
|
网络协议 安全 网络安全
深入解析TURN协议的作用与重要性
【8月更文挑战第24天】
41 0
|
19天前
|
域名解析 网络协议 Linux
在Linux中,我们都知道,dns采用了tcp协议,又采用了udp协议,什么时候采用tcp协议?什么 时候采用udp协议?为什么要这么设计?
在Linux中,我们都知道,dns采用了tcp协议,又采用了udp协议,什么时候采用tcp协议?什么 时候采用udp协议?为什么要这么设计?
|
应用服务中间件 PHP nginx
file not found. nginx php nginx 如何开启解析 PHP 的功能
<p>file not found. nginx php</p> <p>这个问题是你配置文件的问题:</p> <p><br></p> <p><br> 查看就是了不要管<br> nginx 如何开启解析 PHP 的功能?<br> # 成功安装后,创建 php-fpm.conf 配置文件,删除 nginx.conf 中“pass the PHP scripts to FastCGI
6969 0
|
14天前
|
存储 SQL 关系型数据库
PHP与MySQL交互的奥秘
【8月更文挑战第29天】在编程的世界里,PHP和MySQL就像是一对默契的舞伴,共同演绎着数据的交响曲。本文将带你探索它们之间的互动,从连接数据库到执行查询,再到处理结果,每一步都充满了节奏与和谐。我们将一起走进这段代码的旅程,感受数据流动的魅力。

热门文章

最新文章

推荐镜像

更多