indigo-common / indigo-common-php · 提交

转到一个项目

作者 karlet
提交 16b3041e51db884c5cc0baaf7b4a31639883e4d2 1 个父辈 cb67a25a

feat:修复web socket内存溢出

内嵌 并排对比
正在显示 1 个修改的文件 包含 36 行增加 和 38 行删除
<?php
namespace Jiaoyin;
use Swoole\Coroutine;
... ... @@ -7,25 +8,25 @@ use Swoole\WebSocket\Frame;
class Websocket
{
private string $host = '';
private string $path = '';
private int $port = 443;
private bool $ssl = true;
private string $host;
private string $path;
private int $port;
private bool $ssl;
private int $recvTimeout = 60;
public ?Client $client = null;
public string $url = '';
private $lastRecvTime = 0;
private int $lastRecvTime = 0;
private $onOpen = null;
private $onMessage = null;
private $onClose = null;
private $onPing = null;
private $onPong = null;
private $pingState = true;
private $timerCheck = 0;
private $timerPing = 0;
private $desc = 'websocket未命名';
private $pingCount = 0;
private $pingDelay = 10000;//ping tick时间 10s
private bool $pingState = true;
private ?int $timerCheck = 0;
private int $timerPing = 0;
private string $desc = 'websocket未命名';
private int $pingCount = 0;
private int $pingDelay = 10000;//ping tick时间 10s
// 例如：wss://fstream.binance.com/stream、ws://175.178.36.217:9528/spot/stream
public function __construct($url, $desc = null)
... ... @@ -44,10 +45,10 @@ class Websocket
if ($desc) {
$this->desc = 'websocket ' . $desc;
}
output($this->host,$this->port,$this->ssl,$this->path);
output($this->host, $this->port, $this->ssl, $this->path);
}
public function push($data)
public function push($data): void
{
if ($this->client) {
$this->client->push($data);
... ... @@ -56,17 +57,15 @@ class Websocket
}
}
public function close()
public function close(): void
{
swoole_timer_clear($this->timerPing);
swoole_timer_clear($this->timerCheck);
if ($this->client) {
$this->client->close();
}
$this->client?->close();
}
public function connect(callable $onOpen = null, callable $onMessage = null,
callable $onClose = null, $onPing = null, $onPong = null)
callable $onClose = null, $onPing = null, $onPong = null): void
{
$this->close();
$this->pingCount = 0;
... ... @@ -88,8 +87,8 @@ class Websocket
$this->sendPing();
});
while ($this->client) {
$frame = $this->client->recv(60);
if (!$frame && $this->client->errCode!=60) {
$frame = $this->client->recv($this->recvTimeout);
if (!$frame && $this->client->errCode != 60) {
output($this->desc, "错误数据", $frame);
break;
}
... ... @@ -103,7 +102,7 @@ class Websocket
}
if ($frame->opcode == WEBSOCKET_OPCODE_PONG) {
$this->recvPongData($frame->data);
if($onPong){
if ($onPong) {
call_user_func($onPong, $frame->data);
}
}
... ... @@ -115,27 +114,27 @@ class Websocket
break;
}
}
\Swoole\Coroutine::defer(function () use ($onClose) {
output($this->desc, "协程退出");
$this->client = null;
if ($onClose) {
call_user_func($onClose);
}
});
}
Coroutine::defer(function () use ($onClose) {
output($this->desc, "协程退出");
$this->client = null;
if ($onClose) {
call_user_func($onClose);
}
});
} else {
if($onClose){
if ($onClose) {
call_user_func($onClose);
}
output($this->desc, "升级websocket连接失败,1s后重连");
swoole_timer_after(1000,function (){
swoole_timer_after(1000, function () {
$this->connect($this->onOpen, $this->onMessage, $this->onClose, $this->onPing, $this->onPong);
});
}
});
}
static public function createPongData($data = null)
static public function createPongData($data = null): Frame
{
$frame = new Frame();
if ($data) {
... ... @@ -145,7 +144,7 @@ class Websocket
return $frame;
}
static public function createPingData()
static public function createPingData(): Frame
{
$frame = new Frame();
$frame->opcode = WEBSOCKET_OPCODE_PING;
... ... @@ -153,25 +152,24 @@ class Websocket
return $frame;
}
private function sendPing()
private function sendPing(): void
{
$this->push(self::createPingData());
$this->pingState = false;
$this->timerCheck = swoole_timer_after(5000, function () {
if (!$this->pingState) {
//未收到pong且10内未获取数据。假定已断开连接（存在未收到pong,但在传数据的情况。）
if($this->lastRecvTime<time()-10){
if ($this->lastRecvTime < time() - 10) {
output($this->desc, 'ping pong 超时且未收到数据，重新连接');
$this->connect($this->onOpen, $this->onMessage, $this->onClose, $this->onPing, $this->onPong);
}else{
output($this->desc, 'ping pong 超时，再次ping');
} else {
$this->sendPing();
}
}
});
}
private function recvPongData($data)
private function recvPongData($data): void
{
$this->pingState = true;
$this->pingCount += 1;
... ... @@ -181,7 +179,7 @@ class Websocket
swoole_timer_clear($this->timerCheck);
$this->timerPing = swoole_timer_after($this->pingDelay, function () {
//防止重复ping
if($this->pingState){
if ($this->pingState) {
$this->sendPing();
}
});
... ...