MQTT-workerman

###订阅
    <?php
    use Workerman\Worker;
    require_once __DIR__ . '/vendor/autoload.php';
    
    $worker = new Worker();
    $worker->onWorkerStart = function(){
        
        
         // 将db实例存储在全局变量中(也可以存储在某类的静态成员中)
        global $db;
        $db = new \Workerman\MySQL\Connection('127.0.0.1', '3306', 'hjjc', 'eG3CdiKprFm83yX8', 'hjjc');
        
        $mqtt = new Workerman\Mqtt\Client('mqtt://111.222.333.444:1881');
        $mqtt->onConnect = function($mqtt) {
            $mqtt->subscribe('test666');
        };
        $mqtt->onMessage = function($topic, $content) use($db){
            
           // var_dump($topic, $content);
            $json=json_decode($content,true);
       
            $arr=[];
            $arr['topic']=$topic;
            $arr['content']=$content;
    
            var_dump($json);
    
           // 插入
           $insert_id = $db->insert('mqtt')->cols($arr)->query();
           
        };
        $mqtt->connect();
    };
    Worker::runAll();

###发布
    <?php
    use Workerman\Worker;
    require_once __DIR__ . '/vendor/autoload.php';
    
    $worker = new Worker();
    $worker->onWorkerStart = function(){
        $mqtt = new Workerman\Mqtt\Client('mqtt://111.222.333.444:1881');
        $mqtt->onConnect = function($mqtt) {
           $mqtt->publish('test666', 'hello workerman mqtt');
        };
        $mqtt->connect();
    };
    Worker::runAll();

PHP

$data=[];
        $data['瓶11']=rand(1,9);
        $data['瓶21']=rand(1,9);
        $data['瓶31']=rand(1,9);
        $data['瓶41']=rand(1,9);
        $data['flexem_message_id']=time();
        $json=json_encode($data,JSON_UNESCAPED_UNICODE);