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");
// 设置以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);
// 深度校验结果
$result = $this->checksumTest->checksum($data);
if ($result){
print_r(self::getTimestamp()." checksum success\n");
} else {
die(self::getTimestamp()." checksum fail\n");
}
print_r("---------------------------------------------------------------\n");
// 打印增量数据
call_user_func_array($GLOBALS['callback'], array($update));
// 更新全局的全量数据
$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());
}
}
代码块如上,运行信息如下:
workerman运行过程中new Worker创建的新worker实例不会再执行onWorkerStart。
subscribe里不用new worker,直接new AsyncTcpConnection就行,不必用worker包一层。
你这样不断new Worker内存会一直增加。
AsyncTcpConnection记得用完就关闭,不然不断创建又不关闭也会内存泄漏
谢谢大佬,可是在文档里面明确的写了先worker再在onworker中进行客户端连接
而且我现在将wokrer去掉后他启动不了了
谢谢大佬,解决了,是IP问题引起的