toupiaoh5_hyperf/app/Master/Framework/Library/Mqtt/Subscribe.php

<?php
declare(strict_types=1);

namespace App\Master\Framework\Library\Mqtt;

use Hyperf\Coroutine\Coroutine;

class Subscribe{
    /**
     * MQTT 订阅
     * @param array $topic
     * @param $function
     * @return mixed
     */
    public function endlessLoop(array $topic,$function): mixed
    {
        $cline = new MqttClient();
        // 订阅一个主题或者多个主题
        $cline->subscribe($topic);

        //时间
        $timeSincePing = time();

        while (true){

            $function($cline,$topic);

            //心跳
            if ($timeSincePing <= (time() - $cline->getKeepAlive())){
                if ($cline->ping()) {
                    $timeSincePing = time();
                }
            }

            Coroutine::sleep(0.01);
        }
    }
}