From bb56cdb96b7821d0a08274cd4346ead6da437150 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mr=C2=B7x?= Date: Fri, 9 Oct 2020 17:25:57 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=B9=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Kafka/Consumer.php | 3 - Kafka/Consumer/State.php | 8 +- Kafka/Socket.php | 722 +++++++++++++++++++-------------------- 3 files changed, 364 insertions(+), 369 deletions(-) diff --git a/Kafka/Consumer.php b/Kafka/Consumer.php index faa1a635..aaa0b12c 100644 --- a/Kafka/Consumer.php +++ b/Kafka/Consumer.php @@ -73,9 +73,6 @@ class Consumer } $process->start(); $this->isRunning = true; - if ($isBlock) { - \Amp\run(); - } } // }}} diff --git a/Kafka/Consumer/State.php b/Kafka/Consumer/State.php index bfd7850c..a61371f3 100644 --- a/Kafka/Consumer/State.php +++ b/Kafka/Consumer/State.php @@ -136,13 +136,13 @@ class State continue; } $interval = isset($option['interval']) ? $option['interval'] : 200; - \Amp\repeat(function ($watcherId) use ($request, $option) { + Timer::tick($msInterval = $interval, function ($watcherId) use ($request, $option) { if ($this->checkRun($request) && $option['func'] != null) { $context = call_user_func($option['func']); $this->processing($request, $context); } $this->requests[$request]['watcher'] = $watcherId; - }, $msInterval = $interval); + }); } // start sync metadata @@ -150,9 +150,9 @@ class State $context = call_user_func($this->requests[self::REQUEST_METADATA]['func']); $this->processing($request, $context); } - \Amp\repeat(function ($watcherId) { + Timer::tick($msInterval = 1000, function ($watcherId) { $this->report(); - }, $msInterval = 1000); + }); } // }}} diff --git a/Kafka/Socket.php b/Kafka/Socket.php index c1ab358f..e6c56fdd 100644 --- a/Kafka/Socket.php +++ b/Kafka/Socket.php @@ -15,417 +15,415 @@ namespace Kafka; /** -+------------------------------------------------------------------------------ -* Kafka broker socket -+------------------------------------------------------------------------------ -* -* @package -* @version $_SWANBR_VERSION_$ -* @copyright Copyleft -* @author $_SWANBR_AUTHOR_$ -+------------------------------------------------------------------------------ -*/ - + * +------------------------------------------------------------------------------ + * Kafka broker socket + * +------------------------------------------------------------------------------ + * + * @package + * @version $_SWANBR_VERSION_$ + * @copyright Copyleft + * @author $_SWANBR_AUTHOR_$ + * +------------------------------------------------------------------------------ + */ class Socket { - // {{{ consts + // {{{ consts - const READ_MAX_LEN = 5242880; // read socket max length 5MB + const READ_MAX_LEN = 5242880; // read socket max length 5MB - /** - * max write socket buffer - * fixed:send of 8192 bytes failed with errno=11 Resource temporarily - * fixed:'fwrite(): send of ???? bytes failed with errno=35 Resource temporarily unavailable' - * unavailable error info - */ - const MAX_WRITE_BUFFER = 2048; + /** + * max write socket buffer + * fixed:send of 8192 bytes failed with errno=11 Resource temporarily + * fixed:'fwrite(): send of ???? bytes failed with errno=35 Resource temporarily unavailable' + * unavailable error info + */ + const MAX_WRITE_BUFFER = 2048; - // }}} - // {{{ members + // }}} + // {{{ members - /** - * Send timeout in seconds. - * - * @var float - * @access private - */ - private $sendTimeoutSec = 0; + /** + * Send timeout in seconds. + * + * @var float + * @access private + */ + private $sendTimeoutSec = 0; - /** - * Send timeout in microseconds. - * - * @var float - * @access private - */ - private $sendTimeoutUsec = 100000; + /** + * Send timeout in microseconds. + * + * @var float + * @access private + */ + private $sendTimeoutUsec = 100000; - /** - * Recv timeout in seconds - * - * @var float - * @access private - */ - private $recvTimeoutSec = 0; + /** + * Recv timeout in seconds + * + * @var float + * @access private + */ + private $recvTimeoutSec = 0; - /** - * Recv timeout in microseconds - * - * @var float - * @access private - */ - private $recvTimeoutUsec = 750000; + /** + * Recv timeout in microseconds + * + * @var float + * @access private + */ + private $recvTimeoutUsec = 750000; - /** - * Stream resource - * - * @var mixed - * @access private - */ - private $stream = null; + /** + * Stream resource + * + * @var \Swoole\Coroutine\Socket + * @access private + */ + private $stream = null; - /** - * Socket host - * - * @var mixed - * @access private - */ - private $host = null; + /** + * Socket host + * + * @var mixed + * @access private + */ + private $host = null; - /** - * Socket port - * - * @var mixed - * @access private - */ - private $port = -1; + /** + * Socket port + * + * @var mixed + * @access private + */ + private $port = -1; - /** - * Max Write Attempts - * @var int - * @access private - */ - private $maxWriteAttempts = 3; + /** + * Max Write Attempts + * @var int + * @access private + */ + private $maxWriteAttempts = 3; - /** - * Reader watcher - * @var int - * @access private - */ - private $readWatcher = 0; + /** + * Reader watcher + * @var int + * @access private + */ + private $readWatcher = 0; - /** - * Write watcher - * @var int - * @access private - */ - private $writeWatcher = 0; + /** + * Write watcher + * @var int + * @access private + */ + private $writeWatcher = 0; - /** - * Write watcher - * @var int - * @access private - */ - private $writeBuffer = ''; + /** + * Write watcher + * @var int + * @access private + */ + private $writeBuffer = ''; - /** - * Reader buffer - * @var int - * @access private - */ - private $readBuffer = ''; + /** + * Reader buffer + * @var int + * @access private + */ + private $readBuffer = ''; - /** - * Reader need buffer length - * @var int - * @access private - */ - private $readNeedLength = 0; + /** + * Reader need buffer length + * @var int + * @access private + */ + private $readNeedLength = 0; - // }}} - // {{{ functions - // {{{ public function __construct() + // }}} + // {{{ functions + // {{{ public function __construct() - /** - * __construct - * - * @access public - * @param $host - * @param $port - * @param int $recvTimeoutSec - * @param int $recvTimeoutUsec - * @param int $sendTimeoutSec - * @param int $sendTimeoutUsec - */ - public function __construct($host, $port, $recvTimeoutSec = 0, $recvTimeoutUsec = 750000, $sendTimeoutSec = 0, $sendTimeoutUsec = 100000) - { - $this->host = $host; - $this->port = $port; - $this->setRecvTimeoutSec($recvTimeoutSec); - $this->setRecvTimeoutUsec($recvTimeoutUsec); - $this->setSendTimeoutSec($sendTimeoutSec); - $this->setSendTimeoutUsec($sendTimeoutUsec); - } + /** + * __construct + * + * @access public + * @param $host + * @param $port + * @param int $recvTimeoutSec + * @param int $recvTimeoutUsec + * @param int $sendTimeoutSec + * @param int $sendTimeoutUsec + */ + public function __construct($host, $port, $recvTimeoutSec = 0, $recvTimeoutUsec = 750000, $sendTimeoutSec = 0, $sendTimeoutUsec = 100000) + { + $this->host = $host; + $this->port = $port; + $this->setRecvTimeoutSec($recvTimeoutSec); + $this->setRecvTimeoutUsec($recvTimeoutUsec); + $this->setSendTimeoutSec($sendTimeoutSec); + $this->setSendTimeoutUsec($sendTimeoutUsec); + } - /** - * @param float $sendTimeoutSec - */ - public function setSendTimeoutSec($sendTimeoutSec) - { - $this->sendTimeoutSec = $sendTimeoutSec; - } + /** + * @param float $sendTimeoutSec + */ + public function setSendTimeoutSec($sendTimeoutSec) + { + $this->sendTimeoutSec = $sendTimeoutSec; + } - /** - * @param float $sendTimeoutUsec - */ - public function setSendTimeoutUsec($sendTimeoutUsec) - { - $this->sendTimeoutUsec = $sendTimeoutUsec; - } + /** + * @param float $sendTimeoutUsec + */ + public function setSendTimeoutUsec($sendTimeoutUsec) + { + $this->sendTimeoutUsec = $sendTimeoutUsec; + } - /** - * @param float $recvTimeoutSec - */ - public function setRecvTimeoutSec($recvTimeoutSec) - { - $this->recvTimeoutSec = $recvTimeoutSec; - } + /** + * @param float $recvTimeoutSec + */ + public function setRecvTimeoutSec($recvTimeoutSec) + { + $this->recvTimeoutSec = $recvTimeoutSec; + } - /** - * @param float $recvTimeoutUsec - */ - public function setRecvTimeoutUsec($recvTimeoutUsec) - { - $this->recvTimeoutUsec = $recvTimeoutUsec; - } + /** + * @param float $recvTimeoutUsec + */ + public function setRecvTimeoutUsec($recvTimeoutUsec) + { + $this->recvTimeoutUsec = $recvTimeoutUsec; + } - /** - * @param int $number - */ - public function setMaxWriteAttempts($number) - { - $this->maxWriteAttempts = $number; - } + /** + * @param int $number + */ + public function setMaxWriteAttempts($number) + { + $this->maxWriteAttempts = $number; + } - // }}} - // {{{ public function connect() + // }}} + // {{{ public function connect() - /** - * Connects the socket - * - * @access public - * @return void - */ - public function connect() - { - if (!$this->isSocketDead()) { - return; - } + /** + * Connects the socket + * + * @access public + * @return void + */ + public function connect() + { + if (!$this->isSocketDead()) { + return; + } - if (empty($this->host)) { - throw new \Kafka\Exception('Cannot open null host.'); - } - if ($this->port <= 0) { - throw new \Kafka\Exception('Cannot open without port.'); - } + if (empty($this->host)) { + throw new \Kafka\Exception('Cannot open null host.'); + } + if ($this->port <= 0) { + throw new \Kafka\Exception('Cannot open without port.'); + } - $this->stream = @fsockopen( - $this->host, - $this->port, - $errno, - $errstr, - $this->sendTimeoutSec + ($this->sendTimeoutUsec / 1000000) - ); + $this->stream = new \Swoole\Coroutine\Socket(AF_INET, SOCK_STREAM);; - if ($this->stream == false) { - $error = 'Could not connect to ' - . $this->host . ':' . $this->port - . ' ('.$errstr.' ['.$errno.'])'; - throw new \Kafka\Exception($error); - } + if (!$this->stream->connect( + $this->host, + $this->port, + $this->sendTimeoutSec + ($this->sendTimeoutUsec / 1000000) + )) { + $error = 'Could not connect to ' + . $this->host . ':' . $this->port + . ' (' . $this->stream->errMsg . ' [' . $this->stream->errCode . '])'; + throw new \Kafka\Exception($error); + } - stream_set_blocking($this->stream, 0); - stream_set_read_buffer($this->stream, 0); + stream_set_blocking($this->stream, 0); + stream_set_read_buffer($this->stream, 0); - $this->readWatcher = \Amp\onReadable($this->stream, function () { - do { - try { - if(!$this->isSocketDead($this->stream)){ - $newData = @fread($this->stream, self::READ_MAX_LEN); - }else{ - $this->reconnect(); - return; - } - if ($newData) { - $this->read($newData); - } - }catch (\Exception $exception){ - var_dump($exception->getMessage()); - } - } while ($newData); - }); + $this->readWatcher = \Amp\onReadable($this->stream, function () { + do { + try { + if (!$this->isSocketDead($this->stream)) { + $newData = @fread($this->stream, self::READ_MAX_LEN); + } else { + $this->reconnect(); + return; + } + if ($newData) { + $this->read($newData); + } + } catch (\Exception $exception) { + var_dump($exception->getMessage()); + } + } while ($newData); + }); - $this->writeWatcher = \Amp\onWritable($this->stream, function () { - $this->write(); - }, array('enable' => false)); // <-- let's initialize the watcher as "disabled" - } + $this->writeWatcher = \Amp\onWritable($this->stream, function () { + $this->write(); + }, array('enable' => false)); // <-- let's initialize the watcher as "disabled" + } - // }}} - // {{{ public function reconnect() + // }}} + // {{{ public function reconnect() - /** - * reconnect the socket - * - * @access public - * @return void - */ - public function reconnect() - { - $this->close(); - $this->connect(); - } + /** + * reconnect the socket + * + * @access public + * @return void + */ + public function reconnect() + { + $this->close(); + $this->connect(); + } - // }}} - // {{{ public function getSocket() + // }}} + // {{{ public function getSocket() - /** - * get the socket - * - * @access public - * @return void - */ - public function getSocket() - { - return $this->stream; - } + /** + * get the socket + * + * @access public + * @return void + */ + public function getSocket() + { + return $this->stream; + } - // }}} - // {{{ public function SetOnReadable() + // }}} + // {{{ public function SetOnReadable() - public $onReadable; - /** - * set on readable callback function - * - * @access public - * @return void - */ - public function SetOnReadable(\Closure $read) - { - $this->onReadable = $read; - } + public $onReadable; - // }}} - // {{{ public function close() + /** + * set on readable callback function + * + * @access public + * @return void + */ + public function SetOnReadable(\Closure $read) + { + $this->onReadable = $read; + } - /** - * close the socket - * - * @access public - * @return void - */ - public function close() - { - \Amp\cancel($this->readWatcher); - \Amp\cancel($this->writeWatcher); - if (is_resource($this->stream)) { - fclose($this->stream); - } - $this->readBuffer = ''; - $this->writeBuffer = ''; - $this->readNeedLength = 0; - } + // }}} + // {{{ public function close() - /** - * checks if the socket is a valid resource - * - * @access public - * @return boolean - */ - public function isResource() - { - return is_resource($this->stream); - } + /** + * close the socket + * + * @access public + * @return void + */ + public function close() + { + \Amp\cancel($this->readWatcher); + \Amp\cancel($this->writeWatcher); + if (is_resource($this->stream)) { + fclose($this->stream); + } + $this->readBuffer = ''; + $this->writeBuffer = ''; + $this->readNeedLength = 0; + } - // }}} - // {{{ public function read() + /** + * checks if the socket is a valid resource + * + * @access public + * @return boolean + */ + public function isResource() + { + return is_resource($this->stream); + } - /** - * Read from the socket at most $len bytes. - * - * This method will not wait for all the requested data, it will return as - * soon as any data is received. - * - * @param integer $len Maximum number of bytes to read. - * @param boolean $verifyExactLength Throw an exception if the number of read bytes is less than $len - * - * @return string Binary data - * @throws \Kafka\Exception\SocketEOF - */ - public function read($data) - { - $this->readBuffer .= $data; - do { - if ($this->readNeedLength == 0) { // response start - if (strlen($this->readBuffer) < 4) { - return; - } - $dataLen = \Kafka\Protocol\Protocol::unpack(\Kafka\Protocol\Protocol::BIT_B32, substr($this->readBuffer, 0, 4)); - $this->readNeedLength = $dataLen; - $this->readBuffer = substr($this->readBuffer, 4); - } + // }}} + // {{{ public function read() - if (strlen($this->readBuffer) < $this->readNeedLength) { - return; - } - $data = substr($this->readBuffer, 0, $this->readNeedLength); + /** + * Read from the socket at most $len bytes. + * + * This method will not wait for all the requested data, it will return as + * soon as any data is received. + * + * @param integer $len Maximum number of bytes to read. + * @param boolean $verifyExactLength Throw an exception if the number of read bytes is less than $len + * + * @return string Binary data + * @throws \Kafka\Exception\SocketEOF + */ + public function read($data) + { + $this->readBuffer .= $data; + do { + if ($this->readNeedLength == 0) { // response start + if (strlen($this->readBuffer) < 4) { + return; + } + $dataLen = \Kafka\Protocol\Protocol::unpack(\Kafka\Protocol\Protocol::BIT_B32, substr($this->readBuffer, 0, 4)); + $this->readNeedLength = $dataLen; + $this->readBuffer = substr($this->readBuffer, 4); + } - $this->readBuffer = substr($this->readBuffer, $this->readNeedLength); - $this->readNeedLength = 0; - call_user_func($this->onReadable, $data, (int)$this->stream); - } while (strlen($this->readBuffer)); - } + if (strlen($this->readBuffer) < $this->readNeedLength) { + return; + } + $data = substr($this->readBuffer, 0, $this->readNeedLength); - // }}} - // {{{ public function write() + $this->readBuffer = substr($this->readBuffer, $this->readNeedLength); + $this->readNeedLength = 0; + call_user_func($this->onReadable, $data, (int)$this->stream); + } while (strlen($this->readBuffer)); + } - /** - * Write to the socket. - * - * @param string $buf The data to write - * - * @return integer - * @throws \Kafka\Exception\SocketEOF - */ - public function write($data = null) - { - if ($data != null) { - $this->writeBuffer .= $data; - } - $bytesToWrite = strlen($this->writeBuffer); - $bytesWritten = @fwrite($this->stream, $this->writeBuffer); + // }}} + // {{{ public function write() - if ($bytesToWrite === $bytesWritten) { - \Amp\disable($this->writeWatcher); - } elseif ($bytesWritten >= 0) { - \Amp\enable($this->writeWatcher); - } elseif ($this->isSocketDead($this->stream)) { - $this->reconnect(); - } - $this->writeBuffer = substr($this->writeBuffer, $bytesWritten); - } + /** + * Write to the socket. + * + * @param string $buf The data to write + * + * @return integer + * @throws \Kafka\Exception\SocketEOF + */ + public function write($data = null) + { + if ($data != null) { + $this->writeBuffer .= $data; + } + $bytesToWrite = strlen($this->writeBuffer); + $bytesWritten = @fwrite($this->stream, $this->writeBuffer); - // }}} - // {{{ protected function isSocketDead() - - /** - * check the stream is close - * - * @return bool - */ - protected function isSocketDead() - { - return !is_resource($this->stream) || @feof($this->stream); - } + if ($bytesToWrite === $bytesWritten) { + \Amp\disable($this->writeWatcher); + } elseif ($bytesWritten >= 0) { + \Amp\enable($this->writeWatcher); + } elseif ($this->isSocketDead($this->stream)) { + $this->reconnect(); + } + $this->writeBuffer = substr($this->writeBuffer, $bytesWritten); + } - // }}} - // }}} + // }}} + // {{{ protected function isSocketDead() + + /** + * check the stream is close + * + * @return bool + */ + protected function isSocketDead() + { + return !is_resource($this->stream) || @feof($this->stream); + } + + // }}} + // }}} }