php 对接mqtt 完整版本，订阅消息，发送消息

首先打开链接如何在 PHP 项目中使用 MQTT

根据文章让所用依赖安装一下： composer require php-mqtt/client

安装之后弄一个部署

之后在工具里边可以相应链接上

接下来是代码：

 /**** 订阅消息* @return void* @throws \PhpMqtt\Client\Exceptions\ConfigurationInvalidException* @throws \PhpMqtt\Client\Exceptions\ConnectingToBrokerFailedException* @throws \PhpMqtt\Client\Exceptions\DataTransferException* @throws \PhpMqtt\Client\Exceptions\InvalidMessageException* @throws \PhpMqtt\Client\Exceptions\MqttClientException* @throws \PhpMqtt\Client\Exceptions\ProtocolNotSupportedException* @throws \PhpMqtt\Client\Exceptions\ProtocolViolationException* @throws \PhpMqtt\Client\Exceptions\RepositoryException*/public function dingyue(){$server = '*********.aliyun.emqxcloud.cn';//连接地址$port = 1883;$clientId = rand(5, 15);$username = '用户名';$password = '密码';$clean_session = false;$mqtt_version = MqttClient::MQTT_3_1_1;$connectionSettings = (new ConnectionSettings)->setUsername($username)->setPassword($password)->setKeepAliveInterval(60)->setLastWillTopic('emqx/test/last-will')->setLastWillMessage('client disconnect')->setLastWillQualityOfService(1);$mqtt = new MqttClient($server, $port, $clientId, $mqtt_version);$mqtt->connect($connectionSettings, $clean_session);//订阅消息//      下边是轮询订阅，订阅多个设备//        查数据库中设备列表$list = DeviceBall::where(['topic'=>['<>','']])->field('topic')->select();
//        获取topic$list = array_column($list, 'topic');//去重$list = array_unique($list);
//        dump($list);exit;foreach ($list as $i) {$mqtt->subscribe('CIOT/2000/DEVICE/' . $i, function ($topic,$message) use ($mqtt, $i) {printf("Received message on topic [%s]: %s\n", $topic, $message);$message = json_decode($message, true);if (isset($message['Body']['Info']['Msg']['Pass'])) {if ($message['Body']['Info']['Msg']['Pass'] == 1) {
//                        关闭订单$ball_id = DeviceBall::where(['topic' => $i])->find();$order_no = BallOrder::where(['ball_id' => $ball_id['id']])->order('id DESC')->where('status', 1)->value('order_no');$this->confirmorder($order_no);//语音指令，提前关闭，电源关闭$payload = array('Header' => ["Code" => '1004',"Time" => time() . '000',"Sign" => "lvzaina"],'Body' => ["MsgId" => time(),'Cmd' => "Voice",'Index' => 41]);$mqtt->publish(// topic'CIOT/2000/HOST/' . $i,// payloadjson_encode($payload),// qos0,// retaintrue);//可以关闭锁球器//发送关闭指令$payload = array('Header' => ["Code" => '1004',"Time" => time() . '000',"Sign" => "lvzaina"],'Body' => ["Mode" => "0","MsgId" => time(),'Cmd' => "Order",'ID' => 1,'Delay' => 0,'Type' => 0,"DCL" => 0,]);$mqtt->publish(// topic'CIOT/2000/HOST/' . $i,// payloadjson_encode($payload),// qos0,// retaintrue);} else {//异常，播报语音，不关闭锁球器，关锁失败请检车$payload = array('Header' => ["Code" => '1004',"Time" => time() . '000',"Sign" => "lvzaina"],'Body' => ["MsgId" => time(),'Cmd' => "Voice",'Index' => 45]);$mqtt->publish(// topic'CIOT/2000/HOST/' . $i,// payloadjson_encode($payload),// qos0,// retaintrue);}}}, 0);}$mqtt->loop(true);}

发送指令

static function publish(){$server = '********ted.aliyun.emqxcloud.cn';$port = 1883;$clientId = rand(5, 15);$username = '用户名';$password = '密码';$clean_session = false;$mqtt_version = MqttClient::MQTT_3_1_1;$connectionSettings = (new ConnectionSettings)->setUsername($username)->setPassword($password)->setKeepAliveInterval(60)->setLastWillTopic('emqx/test/last-will')->setLastWillMessage('client disconnect')->setLastWillQualityOfService(1);$mqtt = new MqttClient($server, $port, $clientId, $mqtt_version);$mqtt->connect($connectionSettings, $clean_session);$payload = array('Header' => ["Code" => '1004',"Time" => time() . '000',"Sign" => "lvzaina"],'Body' => ["Mode" => "0","MsgId" => time(),'Cmd' => "Check",]);$mqtt->publish(// topic'CIOT/2000/HOST/RE2BBEEC78',//推送的地址// payloadjson_encode($payload),// qos0,// retaintrue);}

上边是代码，下边事最重要的

1首先如果在本地测试的话，新建一个.bat文件，放在根目录

.bat文件内容，然后双击文件执行，这里执行的是上边订阅消息的方法，执行之后，可以在小黑窗里边看到打印的信息

chcp 65001
@echo off
:loop
E:\phpstudy_pro\Extensions\php\php7.4.3nts\php.exe public/index.php api/Billiards/dingyue
ping -n 3 127.1 >nul
goto loop
pause

2，如果是在linux中执行

新建一个.sh文件

.sh文件内容

#!/bin/bash
while true
do# 这里是要执行的代码/www/server/php/72/bin/php public/index.php api/Billiards/dingyue# 这里是要执行的代码endDate=`date +"%Y-%m-%d %H:%M:%S"`echo "At [$endDate] Successful"echo "----------------------------------------------------------------------------"sleep 3
done

开启列队：运行队列的代码

nohup bash run-api.sh >> run-api.out &
然后敲回车敲两下
查看执行文件返回内容
tail -f run-api.out

执行前，要看下是否已经在运行：
ps aux | grep bash

如果已经在执行
结束命令：
kill -9 进程号（256300）

然后再执行ps aux | grep bash 看是否已经停止，重新启动即可。

这样一个.sh文件就启动了