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