station-test-api/task_script/LIVESTOCK_MQTT_FMZL.php

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

use \think\facade\Db;
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 = "livestock_fmzl_mqtt.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 sendData($topic,$data)
{
    $server   = 'stage-agro-mqtt.cacfintech.com';
    $port     = 1883;
    $clientId = 'local_mqtt_livestock_data_cli_123321';
    $username = '';
    $password = "";
    $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);
    rlog('['.date('Y-m-d H:i:s').']connect OK');
    rlog('['.date('Y-m-d H:i:s').']topic:'.$topic);
    $res=$mqtt->publish(
        $topic,
        $data,
        1
    );
    rlog('['.date('Y-m-d H:i:s').']publish end');
    $mqtt->loop(true,true);
    $mqtt->disconnect();
    return $res;
}
$device_arr=[
    '866216067708342-999202300000000','869761079694821-999202300000000','869761079697154-999202300000000','869761079697790-999202300000000','869761079698186-999202300000000','869761079698913-999202300000000','869761079699325-999202300000000','869761079700354-999202300000000','869761079700396-999202300000000','869761079700438-999202300000000','869761079700453-999202300000000','869761079700511-999202300000000','869761079700537-999202300000000','869761079700545-999202300000000','869761079700552-999202300000000','869761079700594-999202300000000','869761079700628-999202300000000','869761079700669-999202300000000','869761079700701-999202300000000','869761079700727-999202300000000','869761079700776-999202300000000','869761079700792-999202300000000','869761079700859-999202300000000','869761079700875-999202300000000','869761079700891-999202300000000','869761079700958-999202300000000','869761079700966-999202300000000','869761079700982-999202300000000','869761079700990-999202300000000','869761079701055-999202300000000','869761079701113-999202300000000','869761079701139-999202300000000','869761079701147-999202300000000','869761079701162-999202300000000','869761079701188-999202300000000','869761079701220-999202300000000','869761079701238-999202300000000','869761079701253-999202300000000','869761079701261-999202300000000','869761079701303-999202300000000','869761079701329-999202300000000','869761079701352-999202300000000','869761079701394-999202300000000','869761079701451-999202300000000','869761079701477-999202300000000','869761079701485-999202300000000','869761079701519-999202300000000','869761079701535-999202300000000','869761079701584-999202300000000','869761079701592-999202300000000','869761079701626-999202300000000','869761079701675-999202300000000','869761079701683-999202300000000','869761079701725-999202300000000','869761079701741-999202300000000','869761079701758-999202300000000','869761079701774-999202300000000','869761079701873-999202300000000','869761079701907-999202300000000','869761079701931-999202300000000','869761079701949-999202300000000','869761079701964-999202300000000','869761079702053-999202300000000','869761079702095-999202300000000','869761079702129-999202300000000','869761079702152-999202300000000','869761079702194-999202300000000','869761079702202-999202300000000','869761079702269-999202300000000','869761079702277-999202300000000','869761079702285-999202300000000','869761079702293-999202300000000','869761079702343-999202300000000','869761079702376-999202300000000','869761079702418-999202300000000','869761079702434-999202300000000','869761079702467-999202300000000','869761079702491-999202300000000','869761079702590-999202300000000','869761079702624-999202300000000','869761079702640-999202300000000','869761079702657-999202300000000','869761079702715-999202300000000','869761079702764-999202300000000','869761079702871-999202300000000','869761079702954-999202300000000','869761079708803-999202300000000','869761079708894-999202300000000','869761079708944-999202300000000','869761079709199-999202300000000','869761079711823-999202300000000'
];

while (1) {
    $jsonData= app_redis()->rpop("livestock_data_after_handle");
    if(!$jsonData){
        sleep(3);
        continue;
    }
    $data=json_decode($jsonData,true);
    if(!in_array($data['deviceId'],$device_arr)){
        continue;
    }
    // var_dump($data);
    rlog("redis data: " . json_encode($data));
    $topic="earings/".$data['deviceId']."/reportData";
    $res=sendData($topic,json_encode($data));

}