適用于協(xié)程客戶端和PHP協(xié)議解析的是什么

這篇文章主要介紹“適用于協(xié)程客戶端和PHP協(xié)議解析的是什么”，在日常操作中，相信很多人在適用于協(xié)程客戶端和PHP協(xié)議解析的是什么問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”適用于協(xié)程客戶端和PHP協(xié)議解析的是什么”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

MQTT是一種基于發(fā)布/訂閱模式的“輕量級”通訊協(xié)議，作為一種低開銷、低帶寬占用的即時通訊協(xié)議，已經(jīng)成為物聯(lián)網(wǎng)的重要組成部分，今天小編就帶大家了解一下simps/mqtt。

MQTT 是一種基于發(fā)布/訂閱（publish/subscribe）模式的"輕量級"通訊協(xié)議，作為一種低開銷、低帶寬占用的即時通訊協(xié)議，已經(jīng)成為物聯(lián)網(wǎng)的重要組成部分

Swoole 也給 PHP 提供了開發(fā)物聯(lián)網(wǎng)項目的能力，只需要設(shè)置一個 open_mqtt_protocol 選項，啟用后就會解析 MQTT 包頭，在 Worker 進程的 onReceive 事件每次都會返回一個完整的 MQTT 數(shù)據(jù)包

當然其他的也有，例如 Workerman 之前提供的 異步 mqtt 客戶端庫 ，還有其他的開源庫，這里就不一一介紹了

Simps 的第一個版本 MQTT 庫 就是參考了 Workerman 的實現(xiàn)，使其能夠使用 Swoole 的協(xié)程能力，同時也修復了一些問題

在此也要感謝 @walkor 對 PHP 生態(tài)作出的貢獻

第一個版本的實現(xiàn)是放在了框架當中，限制了一些用戶的使用。于是又開始了重構(gòu)，將 MQTT 獨立為一個 library ，方便用戶使用的同時也豐富了 PHP 生態(tài)，讓 PHP 程序員不再局限于 Web 開發(fā)

在第一個版本發(fā)布之后，Simps 的交流群中也有不少用戶詢問 MQTT 的問題，Swoole 也修復了一些相關(guān)的 Bug，現(xiàn)在使用 PHP + Swoole 去開發(fā)物聯(lián)網(wǎng)相關(guān)的項目應(yīng)該是如虎添翼

同時第一個版本的 MQTT 庫，只支持 MQTT 3.x，不支持 MQTT 5.0，在 GitHub 上也沒有找到相關(guān)支持的類庫，所以在重構(gòu)了 3.x 版本之后，也支持了一下 MQTT 5.0

也許這是第一個支持 MQTT v5.0 協(xié)議的 PHP library...

支持 MQTT 協(xié)議 3.1、3.1.1 和 5.0 版本，支持 QoS 0、QoS 1、QoS 2，那么它來了，使用 composer 來安裝

composer require simps/mqtt

安裝成功之后我們來看一下訂閱和發(fā)布的使用，以 MQTT5.0 為例

訂閱

首先應(yīng)該是訂閱，訂閱成功之后才能收到對應(yīng)主題的發(fā)布消息，創(chuàng)建一個subscribe.php寫入以下內(nèi)容

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);
    // 訂閱的結(jié)果
    var_dump($res);
    while (true) {
        $buffer = $client->recv();
        if ($buffer && $buffer !== true) {
            $timeSincePing = time();
            // 收到的數(shù)據(jù)包
            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 發(fā)布回復
        if ($buffer['type'] === Types::PUBLISH && $buffer['qos'] === 1) {
            $client->send(
                [
                    'type' => Types::PUBACK,
                    'message_id' => $buffer['message_id'],
                    'code' => ReasonCode::SUCCESS
                ]
            );
        }
    }
});

執(zhí)行php subscribe.php，就會得到這樣的輸出

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

表示訂閱成功，codes 對應(yīng)的是對應(yīng)訂閱主題的 QoS 等級

發(fā)布

訂閱成功之后，創(chuàng)建一個publish.php來測試發(fā)布

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發(fā)布一次消息

打開一個新的終端窗口，執(zhí)行php publish.php就會得到輸出：

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

這里增加了 message，為了用戶可讀，不需要去查找對應(yīng)的 code 含義是什么

返回到訂閱的窗口，就會看到所打印的發(fā)布信息

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)
  }
}

這樣一個簡單的發(fā)布訂閱功能就實現(xiàn)了

在這個庫中還有一些值得優(yōu)化和還未完成的部分，如還沒有支持 MQTT5 的Auth type，以及部分的properties還未支持

想?yún)⑴c的同學可以提交 PR，如果有問題也可以提交 Issue，讓我們共同去建設(shè) PHP 的生態(tài)

到此，關(guān)于“適用于協(xié)程客戶端和PHP協(xié)議解析的是什么”的學習就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關(guān)知識，請繼續(xù)關(guān)注億速云網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>