寫了一段程序。在循環(huán)在redis讀取數(shù)據(jù)，并根據(jù)數(shù)據(jù)發(fā)送主題。mqtt服務器 用的是 Mosquitto
遇到了兩個問題。
1 當qos為2時候發(fā)不出去，0或1的時候就可以。
2 程序運行一段時間 就會提示 Mqtt client: No connection to broker
?
代碼在下面，請看看是哪里用的不對。

<?php

class LabelServer
{
    function __construct(&$task)
    {
        $this->worker = $task;
        $redis = new Redis();
        $this->redis = $redis;
        $redis->connect('127.0.0.1', 6379);
        $mqtt = new \Workerman\Mqtt\Client('mqtt://127.0.0.1:1883', );
        $mqtt->connect();
        $db = \Tools::openDatabase(\Config::$CONFIG, \Config::$CONFIG, \Config::$CONFIG, \Config::$CONFIG);
        $this->db = $db;

        $mqtt->onConnect = function ($mqtt) use ($redis, $task, $db) {

            while (true) {
                while ($redis->LLEN("mqttqueue") > 0) {
                    $value = $redis->RPOP("mqttqueue");
                    if ($value) {
                        Tools::Console("process $value workerid {$task->id}");
                        $avalue = json_decode($value, true);
                        $method = $avalue;
                        switch ($method) {
                            case "device.register":
                                $this->doDeviceRegister($mqtt, $avalue, $avalue);
                                break;
                            default:
                                Tools::Console("doOther $method");
                        }

                    }
                }
                usleep(100000);
            }
        };
    }

    private function doDeviceRegister($mqtt, $topic, $payload)
    {
        Tools::Console("doRegister workerid {$this->worker->id}");
        $insertData = [];
        $insertData = $topic;
        $insertData = json_encode($payload);;
        $this->db->insert("u")->cols($insertData)->query();
        $deviceid = $payload;
        $mqtt->publish("/mtt/shoptalk/$deviceid/device/register_reply", "{ \"id\": \"\", \"version\": \"1.0\", \"params\": { \"code\":200, \"message\":\"注冊成功\" } , \"time\":1524448721000 , \"method\": \"device.register_reply\" }", );
    }

}