修改

task_script/LIVESTOCK_MQTT_CLIENT.php

@@ -214,7 +214,7 @@ function loop()
 
     $server   = '116.62.220.88';
     $port     = 1883;
-    $clientId = 'local_mqtt_livestock_cli_202409271';
+    $clientId = 'local_mqtt_livestock_cli_202410271';
     $username = 'rl517';
     $password = "rlian2022";
 

task_script/PUMP_MQTT_PUBLISH.php

@@ -0,0 +1,477 @@
+<?php
+require('../vendor/autoload.php');
+use \PhpMqtt\Client\MqttClient;
+use \PhpMqtt\Client\ConnectionSettings;
+use think\facade\Cache;
+// date_default_timezone_set("America/Bahia");
+// define('HOST', '127.0.0.1');
+// define('PORT', '6379');
+// define('PASSWORD', '123456');
+// define('DATABASE', 2);
+define('HOST', 'r-bp1eebab79320044pd.redis.rds.aliyuncs.com');
+define('PORT', '6379');
+define('PASSWORD', '7e2b5c91e438be3c!');
+define('DATABASE', 4);
+
+function app_redis()
+{
+    static $redis = null;
+    static $conn = false;
+    if (!$conn) {
+        connect: //定义标签
+        $redis = new Redis();
+        try {
+            //建立的Redis短连接,在请求结束后不会自动关闭,相当于持久连接.
+            $conn = $redis->connect(HOST, PORT);
+            $conn = $redis->auth(PASSWORD);
+            $conn = $redis->select(DATABASE);
+            // 连接成功,返回$redis对象,连接失败,返回false.
+            return ($conn === true) ? $redis : false;
+        } catch (Exception $e) {
+            return false;
+        }
+    } else {
+        // 这里假设PHP-FPM在处理一个请求的时间内,Redis连接都是可用的.
+        // 所以只在PHP-CLI下检查Redis连接的状态,进行断线重连.
+        if (php_sapi_name() === 'cli') {
+            try {
+                // ping用于检查当前连接的状态,成功时返回+PONG,失败时抛出一个RedisException对象.
+                // ping失败时警告:
+                // Warning: Redis::ping(): connect() failed: Connection refused
+                // var_dump('AAAAAAAAA', $redis);
+                // echo 'Redis 连接状态' . $redis->ping() . PHP_EOL;
+                @$redis->ping();
+                if (!$redis->ping()) {
+                    goto connect; //跳转到标签出继续执行连接操作
+                }
+            } catch (Exception  $e) {
+                // 信息如 Connection lost 或 Redis server went away
+                echo $e->getMessage();
+                echo 'Redis 连接失败 重新连接：' . PHP_EOL;
+                // 断线重连
+                goto connect;
+            }
+        }
+        return $redis;
+    }
+}
+
+
+function sendConfig($topic,$config)
+{
+    $server   = '116.62.220.88';
+    $port     = 1883;
+    $clientId = 'local_mqtt_pump_config';
+    $username = 'rl517';
+    $password = "rlian2022";
+    $clean_session = false;
+
+    $connectionSettings  = new ConnectionSettings();
+    $connectionSettings = $connectionSettings
+        ->setUsername($username)
+        ->setPassword($password)
+        ->setKeepAliveInterval(60)
+        // Last Will 设置
+        //        ->setLastWillTopic('emqx/test/last-will')
+        //        ->setLastWillMessage('client disconnect')
+        //        ->setLastWillQualityOfService(1)
+    ;
+
+    $mqtt = new MqttClient($server, $port, $clientId);
+
+    $mqtt->connect($connectionSettings, $clean_session);
+    echo 'connect OK'.PHP_EOL;
+    echo 'topic:'.$topic.PHP_EOL;
+    echo 'config:'.$config.PHP_EOL;
+    $res=$mqtt->publish(
+        $topic,
+        $config,
+        1
+    );
+    echo 'publish end'.PHP_EOL;
+    $mqtt->loop(true,true);
+    $mqtt->disconnect();
+    return $res;
+}
+function updateSendResult($msgid){
+    $conn = new mysqli('rm-bp1h3uqkzy66ckt8y125010.mysql.rds.aliyuncs.com', 'dev', '711e7D69f9d0c3f1', 'catch_pump');
+    // $conn = new mysqli('127.0.0.1', 'root', 'root', 'catch_pump');
+    if ($conn -> connect_errno) {
+        printf("Connect failed: %s\n", $conn->connect_error);
+        exit();
+    }
+    $sql = "UPDATE send_config_log SET result='1' WHERE id=".$msgid;
+    if ($conn->query($sql) === TRUE) {
+        echo "send_config_log update success";
+    } else {
+        echo "更新失败: " . $conn->error;
+    }
+
+}
+
+function rlog(...$args)
+{
+    //函数参数是可变参数 使用前查下static开始的大写变量配置是否正确
+    //rlog('info', [1,2,3], $a = null, '[]', $this, false);
+    $args = func_get_args();
+//        if (empty($args[0])) {
+//            return;
+//        }
+    static $LOG_CONSOLE = true;//是否输出到控制台 fpm需要为false cli可以为true
+
+    if (!empty(getenv('PM2_HOME'))) { //pm2
+        $LOG_CONSOLE = false;
+    } else if (PHP_SAPI == 'fpm-fcgi') { //php-fpm
+        $LOG_CONSOLE = false;
+    } else if (PHP_SAPI == 'cgi')  { //cli
+        $LOG_CONSOLE = true;
+    } else if (PHP_SAPI == 'cli-server')  { //php think run
+        $LOG_CONSOLE = false;
+    } else {
+        $LOG_CONSOLE = true;
+    }
+
+    static $LOG_NAME = "./pubsub_tcp_iiot.log";//值为空时 不写入文件 第一个参数是imlog时也可以更改
+    if (stripos(PHP_OS, 'WIN') !== false) {
+        $LOG_NAME = "C:\Windows\Temp\php.log";
+    }
+    static $LOG_SIZE = 128 * 1024 * 1024;//文件最大尺寸 大于这个尺寸时 会生成个后缀.old的文件
+
+    static $LOG_CACHE = false;//是否缓存日志内容 用于批量写入文件 如需强制刷新第一个参数传sync
+    static $CACHE_DURATION = 10;//缓存最大时间 秒
+    static $CACHE_SIZE = 1024;//缓存大小
+    static $cacheStartTime = 0;
+    static $cacheBuf = '';
+
+    static $LOG_TIMES = 10;//调用这个函数最大次数 超过次数或$logCount==1 会判断下文件大小 决定是否新生成文件
+    static $logCount = 0;
+
+    static $MAX_LEN = 1024 * 4;//数据不能超过这个  不然就截断了
+
+    $sync = false;//如果是true 使用$LOG_CACHE时会把数据保存到磁盘
+
+    $implicit0 = ['imsync', 'imtrace', 'imlog'];//这个函数的参数0 隐藏用法
+
+    $buf = '';
+    if (count($args) == 1 && $args[0] === "\n") {//只有换行时 不写入时间戳了
+        $buf = "\n";
+    } else {
+        $sync = is_string($args[0]) && strtolower($args[0]) == $implicit0[0];
+
+        if (is_string($args[0]) && strpos($args[0], $implicit0[2]) === 0) {//设置日志名
+            $LOG_NAME = str_replace($implicit0[2], '', $args[0]);
+//                unset($args[0]);//记入日志也可以
+        } else {
+//            if ($LOG_NAME == 'php.log') {//每个文件会生成不同的文件名
+//                $LOG_NAME = str_replace(".php", '', basename(__FILE__)) . '.log';
+//            }
+        }
+
+        $pid = '';//进程id
+        if (function_exists('posix_getpid')) {
+            $pid = ' ' . posix_getpid() . ' ';
+        }
+        $fileLine = '';//文件名:行号
+        {
+            $debug = debug_backtrace();
+            $backtrace = 0;
+            if (is_string($args[0]) && strpos($args[0], $implicit0[1]) === 0) {
+                $backtrace = intval(str_replace($implicit0[1], '', $args[0]));
+                unset($args[0]);
+            } else if ($sync) {
+                unset($args[0]);
+            }
+            $fileLine = ($pid == '' ? ' ' : '') . str_replace(".php", '', basename($debug[$backtrace]['file']))
+                . ':' . $debug[$backtrace]['line'] . '';
+
+        }
+
+        $allPara = '';
+        foreach ($args as $para) {
+            if (is_array($para)) {
+                $allPara .= json_encode($para) . ' ';
+            } else if (is_object($para)) {
+                if (method_exists($para, '__toString')) {
+                    $allPara .= $para . ' ';
+                } else {
+                    $allPara .= get_class($para) . json_encode($para) . ' ';
+                }
+            } else if (is_bool($para)) {
+                $allPara .= $para ? 'true ' : 'false ';
+            } else if (is_null($para)) {
+                $allPara .= 'null ';
+            } else {
+                $allPara .= $para . ' ';
+            }
+        }
+
+        $len = strlen($allPara);
+        if ($len > $MAX_LEN) {
+            $allPara = substr($allPara, 0, $MAX_LEN) . "({$len})......";
+        }
+
+        $buf = "[" . date("y-m-d H:i:s.") . substr(microtime(FALSE), 2, 3) . "{$pid}{$fileLine}]" . $allPara . "\n";
+    }
+
+    $logCount++;
+    if (!empty($LOG_NAME)) {
+        if ($LOG_CACHE) {
+            if ($cacheBuf == '') {
+                $cacheStartTime = time();
+            }
+            $cacheBuf .= $buf;
+            //超过缓存尺寸 或者 超过缓存时长 写缓存到文件
+            if (strlen($cacheBuf) > $CACHE_SIZE || time() - $cacheStartTime > $CACHE_DURATION) {
+                $cacheStartTime = time();
+                goto write;
+            } else {
+                if ($sync) {
+                    goto write;
+                } else {
+                    goto skipWrite;
+                }
+            }
+        } else {
+            $cacheBuf = $buf;
+        }
+        write: {
+            if ($logCount > 100) {//会有缓存
+                clearstatcache();
+            }
+            //超过尺寸后 删除旧文件 把新文件重命名为旧文件
+            if (($logCount == 1 || $logCount > $LOG_TIMES)
+                && file_exists($LOG_NAME) && filesize($LOG_NAME) > $LOG_SIZE) {
+
+                //获取独占锁
+                $fp = fopen($LOG_NAME . '.lock', 'a');
+                $k = 0;
+                do {
+                    $isLock = flocK($fp, LOCK_EX);
+                    $k++;
+                    if (!$isLock && $k > 1000) {
+                        echo "lock 1000\n";
+                        goto PUT;
+                    }
+                } while (!$isLock);
+
+                //另外一个进程进入时 重新判断下
+                clearstatcache();
+                if (filesize($LOG_NAME) <= $LOG_SIZE) {
+                    goto UN;
+                }
+
+                //删除旧文件和重命名文件
+                $oldLogName = $LOG_NAME . '.old';
+                if (file_exists($oldLogName)) {
+                    if (!unlink($oldLogName)) {
+                        echo "unlink err\n";
+                    }
+                }
+                if (!rename($LOG_NAME, $oldLogName)) {
+                    echo "rename err\n";
+                }
+
+                //解锁
+                UN:
+                flock($fp, LOCK_UN);
+                fclose($fp);
+
+
+                $logCount = 0;
+            }
+            PUT:
+            if (!file_put_contents($LOG_NAME, $cacheBuf, FILE_APPEND)) {
+                echo "file_put_contents err," . getcwd() . "/{$LOG_NAME} \n";
+            }
+            $cacheBuf = '';
+        }
+        skipWrite:{
+        }
+    }
+    if ($LOG_CONSOLE) {
+        echo $buf;
+    }
+}
+function printChars($str) {
+    $result = '';
+    $length = strlen($str);
+    for ($i = 0; $i < $length; $i++) {
+        $char = $str[$i];
+        $ordValue = ord($char);
+
+        // 如果是控制字符（ASCII 值小于32 或者 是换行、回车、制表符）
+        if ($ordValue >= 32 && $ordValue <= 126) {
+
+            $result .= $char;
+        } else {
+            // 对于其他字符直接输出
+            $result .= '.';
+//            switch ($ordValue) {
+//                case 9: // 制表符
+//                    $result .= '\t';
+//                    break;
+//                case 10: // 换行符
+//                    $result .= '\n';
+//                    break;
+//                case 13: // 回车符
+//                    $result .= '\r';
+//                    break;
+//                default:
+//                    // 其他控制字符可以使用其 ASCII 值的十六进制形式
+//                    $result .= '\\x' . dechex($ordValue);
+//                    break;
+//            }
+        }
+    }
+    return $result;
+}
+
+
+class MQTT_TYPE {/*对应从1开始*/
+    public static $PUB_MEASURE = 'a'; //设备->平台 设备上报 的压力值     /*1*/
+    public static $SUB_ALARM = 'b';   //平台->设备 平台下发 的告警值
+    public static $SUB_ALARM_REPLAY = 'c'; //设备->平台 设备上报 对平台下发的告警值的回复
+    public static $GET_ALARM = 'd'; //平台下发 获取设备告警值
+
+    //设备->平台 设备上报 下面6个 顺序不要变 数据库里面存的是1-6 本页面insertAlarm函数有ord转对应的数字
+    public static $PUB_UPPER_ALARM_START_EVENT = 'e';       /*5*/
+    public static $PUB_UPPER_ALARM_STOP_EVENT = 'f';
+    public static $PUB_LOWER_ALARM_START_EVENT = 'g';
+    public static $PUB_LOWER_ALARM_STOP_EVENT = 'h';
+    public static $PUB_UPPER_ALARM_SETTING_CHANGE = 'i';
+    public static $PUB_LOWER_ALARM_SETTING_CHANGE = 'j';      /*10*/
+
+    public static $UP_ALARM = 'k';//设备->平台
+
+    //平台->设备 校准0
+    public static $PUB_CALIBRATION_ZERO = 'l';
+    //设备->平台 校准0回复
+    public static $PUB_CALIBRATION_ZERO_REPLAY = 'm';
+}
+
+define("ALTER_SEP", ",");
+//成功返回二进制, 失败返回false
+function buildMqttData($mqttType, $msgId/*整数 0-4294967295*/, $data) {
+    /**test**/
+//    $mqttType = MQTT_TYPE::$SUB_ALARM;
+//    $data['lower'] = '12.34';
+//    $data['upper'] = '56.789';
+//
+//    $mqttType = MQTT_TYPE::$GET_ALARM;
+    /**test**/
+
+    rlog("build", $mqttType, $msgId, $data);
+
+    $msgId = intval($msgId);
+    if ($msgId < 0 || $msgId > 0xffffffff) {
+        rlog("err msgId");
+        return false;
+    }
+    $MIN = 0;
+    $MAX = 39.99;
+    $ALERT_LEN = 6;//值最大
+
+    $bin = pack("C1N1", ord($mqttType), $msgId/*time()*/ /*这个时间大小端和设备端相反*/);
+
+    if ($mqttType == MQTT_TYPE::$SUB_ALARM) {
+        /*
+         * 可以单独设置一个值,不需要设置的值 可像下面这样置空
+         *  $data['lower'] = ''
+         * */
+        if (!isset($data['lower']) || strlen($data['lower']) >= $ALERT_LEN) {
+            rlog("lower val err, max len " . $ALERT_LEN);
+            return false;
+        }
+        if (!isset($data['upper']) || strlen($data['upper']) >= $ALERT_LEN) {
+            rlog("upper val err, max len " . $ALERT_LEN);
+            return false;
+        }
+
+        $data['lower'] = trim($data['lower']);
+        $data['upper'] = trim($data['upper']);
+        if ($data['lower'] == '' && $data['upper'] == '') {
+            rlog("val empty");
+            return false;
+        }
+
+        //0-39.999 值范围
+        //char str[] = "23.556,44.25";//设置下限和上限
+        //char str[] = " ,44.25";     //设置上限 前面有个空格
+        //char str[] = "23.556,";      //设置下限
+        if ($data['lower'] == '') {
+            $data['lower'] = ' ';//要有个空格
+        } else {
+            if (!preg_match('/^-?\d+(\.\d+)?$/', $data['lower'])) {
+                rlog("err lower " . $data['lower']);
+                return false;
+            } else {
+                if (floatval($data['lower']) < $MIN || floatval($data['lower']) > $MAX) {
+                    rlog("err lower range " . $data['lower']);
+                    return false;
+                }
+            }
+        }
+        if (!empty($data['upper'])) {
+            if (!preg_match('/^-?\d+(\.\d+)?$/', $data['upper'])) {
+                rlog("err upper " . $data['upper']);
+                return false;
+            } else {
+                if (floatval($data['upper']) < $MIN || floatval($data['upper']) > $MAX) {
+                    rlog("err upper range " . $data['upper']);
+                    return false;
+                }
+            }
+        }
+
+
+        $buff = $data['lower'] . ALTER_SEP . $data['upper'];/*1.236,10.256*/
+        $bin = $bin . $buff . chr(0);//补零 设备解析是以为\0结尾
+    } else if ($mqttType == MQTT_TYPE::$GET_ALARM) {
+    } else if ($mqttType == MQTT_TYPE::$PUB_CALIBRATION_ZERO) {
+    } else {
+        rlog("warn unknown type " . ord($mqttType));
+        return false;
+    }
+    rlog('build ' . bin2hex($bin) . '(' . printChars($bin) . ')');
+    return $bin;
+}
+
+
+
+
+  
+
+while (1) {
+    $jsonData= app_redis()->rpop("pump_mqtt_config_list");
+    if(!$jsonData){
+        sleep(3);
+        continue;
+    }
+    $data=json_decode($jsonData,true);
+    var_dump($data);
+
+    $topic="DEV/DATA/".$data['mac'];
+    $config_json=$data['config'];
+    $sendData=false;
+
+    if($data['type']=='config'){
+        $sendData = buildMqttData(MQTT_TYPE::$SUB_ALARM, $config_json['msgId'], ['lower' =>$config_json['lower'], 'upper' =>$config_json['upper']]);
+    }elseif($data['type']=='alarm'){
+        $sendData = buildMqttData(MQTT_TYPE::$GET_ALARM, $config_json['msgId'], null);
+    }elseif($data['type']=='calibration'){
+        $sendData = buildMqttData(MQTT_TYPE::$PUB_CALIBRATION_ZERO, $config_json['msgId'], null);
+    }else{
+
+    }
+    var_dump($sendData);
+        
+    if($sendData){
+        updateSendResult($config_json['msgId']);
+        $res=sendConfig($topic,$sendData);
+    }
+    
+    
+    
+    
+
+}