function subscribe($callback, $sub_str="") {
        $GLOBALS['sub_str'] = $sub_str;
        $GLOBALS['callback'] = $callback;
        $worker = new Worker();
        if(isset(Config::$config['paper'])){
            $url = "";
        }else{
            $url = "";
        }

        try {
            $worker->onWorkerStart = function($worker) use ($url){
                // ssl需要訪問443端口
                $con = new AsyncTcpConnection($url);

                $ntime = $this->getTimestamp();
                print_r($ntime." $url\n");

                // 設(shè)置以ssl加密方式訪問，使之成為wss
                $con->transport = 'ssl';

                // 定時器
                Timer::add(20, function() use ($con)
                {
                    $con->send("ping");
//
                    $ntime = $this->getTimestamp();
                    print_r($ntime." ping\n");
                });

                $con->onConnect = function($con){
                    $data = json_encode([
                        'op' => "subscribe",
//                        'args' => $GLOBALS['sub_str']
                        'args' => [
                            $GLOBALS['sub_str']
                        ]
                    ]);

                    $data = stripslashes($data);
//                    $data = substr($data,1,strlen());
                    $data = '{"op":"subscribe","args":[{'.substr($data,28,-4).'}]}';
                    $ntime = $this->getTimestamp();
                    print_r($ntime . " $data\n");

                    $con->send($data);
//                    $con->send($data);
                };

                $con->onMessage = function($con, $data) {

                    // 如果是深度200檔，則校驗
                    if(strpos($data,"checksum"))
                    {
                        $ntime = $this->getTimestamp();
                        print_r($ntime . " $data\n");

                        if ($this->partial==null)
                        {
                            $this->partial=$data;
                        }else{
                            $update = $data;
                            // 深度合并
                            $data = $this->checksumTest->depthMerge($this->partial,$update);
                            // 深度校驗結(jié)果
                            $result = $this->checksumTest->checksum($data);

                            if ($result){
                                print_r(self::getTimestamp()." checksum success\n");
                            } else {
                                die(self::getTimestamp()." checksum fail\n");
                            }

                            print_r("---------------------------------------------------------------\n");
                            // 打印增量數(shù)據(jù)
                            call_user_func_array($GLOBALS['callback'], array($update));
                            // 更新全局的全量數(shù)據(jù)
                            $this->partial = $data;
                        }
                    }else{
                        $ntime = $this->getTimestamp();
                        print_r($ntime . " $data\n");
                    }
                };

                $con->onClose = function ($con) {
                    $ntime = $this->getTimestamp();
                    print_r($ntime." reconnecting\n");
                    $con->reConnect(0);
                };

                $con->connect();
            };

            Worker::runAll();
        } catch (\Exception $e) {
            var_dump($e->getMessage());
        }
    }

代碼塊如上，運行信息如下：