|
|
@@ -0,0 +1,253 @@
|
|
|
+<?php
|
|
|
+require('../vendor/autoload.php');
|
|
|
+use \PhpMqtt\Client\MqttClient;
|
|
|
+use \PhpMqtt\Client\ConnectionSettings;
|
|
|
+use think\facade\Cache;
|
|
|
+date_default_timezone_set("PRC");
|
|
|
+define('HOST', '127.0.0.1');
|
|
|
+define('PORT', '6379');
|
|
|
+define('PASSWORD', '123456');
|
|
|
+define('DATABASE', 3);
|
|
|
+define('OFFLINETIMEINT', 120);
|
|
|
+
|
|
|
+
|
|
|
+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 rlog(...$args)
|
|
|
+{
|
|
|
+ if (empty($args[0])) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ static $LOG_CONSOLE = false; //是否输出到控制台
|
|
|
+ static $LOG_NAME = "parea_status_update.log"; //值为空时 不写入文件
|
|
|
+ static $LOG_SIZE = 64 * 1024 * 1024; //文件最大尺寸
|
|
|
+
|
|
|
+ static $LOG_CACHE = false; //是否缓存日志内容 用于批量写入文件
|
|
|
+ static $CACHE_DURATION = 10; //缓存最大时间 秒
|
|
|
+ static $CACHE_SIZE = 1024; //缓存大小
|
|
|
+ static $cacheStartTime = 0;
|
|
|
+ static $cacheBuf = '';
|
|
|
+
|
|
|
+ static $LOG_TIMES = 10; //调用这个函数最大次数 超过次数后判断下文件大小
|
|
|
+ static $logCount = 0;
|
|
|
+
|
|
|
+
|
|
|
+ $buf = '';
|
|
|
+ if (count($args) == 1 && $args[0] == "\n") { //只有换行时 不写入时间戳了
|
|
|
+ $buf = "\n";
|
|
|
+ } else {
|
|
|
+ $pid = ''; //进程id
|
|
|
+ if (function_exists('posix_getpid')) {
|
|
|
+ $pid = ' ' . posix_getpid() . ' ';
|
|
|
+ }
|
|
|
+ $fileLine = ''; //文件名:行号
|
|
|
+ {
|
|
|
+ $debug = debug_backtrace();
|
|
|
+ $fileLine = ($pid == '' ? ' ' : '') . basename($debug[0]['file']) . ':' . $debug[0]['line'] . ' ';
|
|
|
+ }
|
|
|
+ $buf = date("y-m-d H:i:s") . "{$pid}{$fileLine}" . implode(' ', $args) . "\n";
|
|
|
+ }
|
|
|
+
|
|
|
+ $logCount++;
|
|
|
+ if (!empty($LOG_NAME)) {
|
|
|
+ if ($LOG_CACHE) {
|
|
|
+ $cacheBuf .= $buf;
|
|
|
+ //超过缓存尺寸 或者 超过缓存时长 写缓存到文件
|
|
|
+ if (strlen($cacheBuf) > $CACHE_SIZE || time() - $cacheStartTime > $CACHE_DURATION) {
|
|
|
+ $cacheStartTime = time();
|
|
|
+ goto write;
|
|
|
+ } else {
|
|
|
+ goto skipWrite;
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ $cacheBuf = $buf;
|
|
|
+ }
|
|
|
+ write: {
|
|
|
+ //超过尺寸后 删除旧文件 把新文件重命名为旧文件 多进程同时操作 不加锁问题不大
|
|
|
+ if ($logCount > $LOG_TIMES && filesize($LOG_NAME) > $LOG_SIZE) {
|
|
|
+ $oldLogName = $LOG_NAME . '.old';
|
|
|
+ if (file_exists($oldLogName)) {
|
|
|
+ if (!unlink($oldLogName)) {
|
|
|
+ echo "unlink err\n";
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (!rename($LOG_NAME, $oldLogName)) {
|
|
|
+ echo "rename err\n";
|
|
|
+ }
|
|
|
+ $logCount = 0;
|
|
|
+ }
|
|
|
+ if (!file_put_contents($LOG_NAME, $cacheBuf, FILE_APPEND)) {
|
|
|
+ echo "file_put_contents err\n";
|
|
|
+ }
|
|
|
+ $cacheBuf = '';
|
|
|
+ }
|
|
|
+ skipWrite: {
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if ($LOG_CONSOLE) {
|
|
|
+ echo $buf;
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+function http($url, $params, $header = [], $method = 'GET', $timeout = 10)
|
|
|
+{
|
|
|
+ // POST $params 字符串形式query=abcd&abc=12345
|
|
|
+ //GET $params 数组['query' => 'abcd', 'abc' => 12345]
|
|
|
+ // $header[] = "Content-Type: application/x-www-form-urlencoded";
|
|
|
+ // $header[] = "Content-Type: application/soap+xml; charset=utf-8";
|
|
|
+ // $header[] = "Content-Type: application/json; charset=utf-8";
|
|
|
+ // $header[] = "Expect: ";
|
|
|
+ rlog("[HTTP] url:$url,method:$method" . ",header:" . json_encode($header));
|
|
|
+ if (strtoupper($method) == 'POST') {
|
|
|
+ rlog("[POST] send params " . (!is_array($params) ? $params : json_encode($params, JSON_UNESCAPED_UNICODE)));
|
|
|
+ } else {
|
|
|
+ rlog("[GET] send " . json_encode($params));
|
|
|
+ }
|
|
|
+ $header[] = "Expect: ";
|
|
|
+ $opts = array(
|
|
|
+ CURLOPT_TIMEOUT => $timeout,
|
|
|
+ CURLOPT_RETURNTRANSFER => 1,
|
|
|
+ CURLOPT_SSL_VERIFYPEER => false,
|
|
|
+ CURLOPT_SSL_VERIFYHOST => false,
|
|
|
+ CURLOPT_HTTPHEADER => $header
|
|
|
+ );
|
|
|
+
|
|
|
+ /* 根据请求类型设置特定参数 */
|
|
|
+ switch (strtoupper($method)) {
|
|
|
+ case 'GET':
|
|
|
+ $opts[CURLOPT_URL] = $url . (empty($params) ? '' : ('?' . http_build_query($params)));
|
|
|
+ break;
|
|
|
+ case 'POST':
|
|
|
+ //$params = http_build_query($params);
|
|
|
+ $opts[CURLOPT_URL] = $url;
|
|
|
+ $opts[CURLOPT_POST] = 1;
|
|
|
+ $opts[CURLOPT_POSTFIELDS] = json_encode($params);
|
|
|
+ break;
|
|
|
+ default:
|
|
|
+ rlog("[ERR] method " . $method);
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+
|
|
|
+ global $ch; //curl长连接
|
|
|
+ if (empty($ch)) {
|
|
|
+ $ch = curl_init();
|
|
|
+ }
|
|
|
+ if (empty($ch)) {
|
|
|
+ rlog("[ERR] curl_init");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+
|
|
|
+ $csa = curl_setopt_array($ch, $opts);
|
|
|
+ if (empty($csa)) {
|
|
|
+ rlog("[ERR] curl_setopt_array");
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+
|
|
|
+ $data = curl_exec($ch);
|
|
|
+ if ($data === false) {
|
|
|
+ rlog("[ERR] curl_exec errno:" . curl_errno($ch) . " " . curl_error($ch));
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ //unicode转中文
|
|
|
+ $data = decodeUnicode($data);
|
|
|
+ rlog("[HTTP] recv " . $data);
|
|
|
+ //curl_close($ch);
|
|
|
+ return $data;
|
|
|
+}
|
|
|
+function decodeUnicode($str)
|
|
|
+{
|
|
|
+ return preg_replace_callback('/\\\\u([0-9a-f]{4})/i', function ($matches) {
|
|
|
+ return iconv("UCS-2BE", "UTF-8", pack("H*", $matches[1]));
|
|
|
+ }, $str);
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+function mqttToRedis($text){
|
|
|
+
|
|
|
+ try{
|
|
|
+ app_redis()->lpush("mqtt_data_parea4rssi",$text);
|
|
|
+ }catch(\Exception $e){
|
|
|
+ rlog("INFO", 'recv',"redis 异常".$e->getMessage());
|
|
|
+ }
|
|
|
+
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+while (true) {
|
|
|
+ try {
|
|
|
+ rlog('INFO', 'task start');
|
|
|
+
|
|
|
+ $infos = app_redis()->hgetall("parea_rfidinfos");
|
|
|
+
|
|
|
+ foreach ($infos as $k => $v) {
|
|
|
+ # code...
|
|
|
+ $data = json_decode($v, true);
|
|
|
+ if ( (time() - $data["time"] ) > OFFLINETIMEINT ) {
|
|
|
+ # code...
|
|
|
+ if ($data["status"] == 1) {
|
|
|
+ # code...
|
|
|
+ $url="http://47.114.185.186:8115/api/areaReport";
|
|
|
+ $url_data = [
|
|
|
+ "mac" => $data["mac"],
|
|
|
+ "data" => [
|
|
|
+ [
|
|
|
+ "label" => $data["id"],
|
|
|
+ "time" => $data["time"],
|
|
|
+ "dirt" => 2
|
|
|
+ ]
|
|
|
+ ]
|
|
|
+ ];
|
|
|
+ http($url, $url_data, [], 'POST');
|
|
|
+ }
|
|
|
+ app_redis()->hdel('parea_rfidinfos',$k);;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ } catch (\Exception $ex) {
|
|
|
+ rlog("INFO", 'pushhttp',"推送 异常".$ex->getMessage());
|
|
|
+ }
|
|
|
+ sleep(60); //1分钟遍历一次
|
|
|
+}
|
|
|
+
|
