station-test-api/task_script/LIVESTOCK_MQTT_PUBLISH.php

<?php
require('../vendor/autoload.php');
use \PhpMqtt\Client\MqttClient;
use \PhpMqtt\Client\ConnectionSettings;
use think\facade\Cache;
date_default_timezone_set("PRC");
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 sendConfig($topic,$config)
{
    $server   = '116.62.220.88';
    $port     = 1883;
    $clientId = 'mqtt_livestock_config_cli';
    $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;
    $res=$mqtt->publish(
        $topic,
        $config,
        0
    );
    $mqtt->disconnect();
    return $res;
}


$redis=app_redis();
while (1) {
    $jsonData= $redis->rpop("device_mqtt_config_list");
    if(!$jsonData){
        sleep(3);
        continue;
    }
    $data=json_decode($jsonData,true);
    $topic="earings/".$data['device_id']."/coludControl";
    $config=[];
    if($data['type']=='exe'){
        $config=[
            $data['type']=>$data['config']
        ];
    }else{
        $config=[
            $data['type']=>[
                "config"=>$data['config']
            ]
        ];
    }
   
    $config_json=json_encode($config);
    var_dump($config_json);
    $res=sendConfig($topic,$config_json);
    if($res){
        Db::table('send_config_log')->where('id',$data['msgid'])->update(['result'=>'1']);
    }
    var_dump($res);
}