基於PHPstream擴充套件手動實現一個redis客戶端
阿新 • • 發佈：2020-08-28
描述
redis是一個經典的key-value快取資料庫，採用C/S架構。當我們安裝成功以後，你就知道它有個服務端，啟動後預設監聽6379埠，然後提供一個客戶端工具redis-cli。
我們可以使用redis-cli然後書寫命令與服務端通訊。
上面我們大概知道了redis的工作模式，為了更好的認知它，我就開始思考如何自己去連線服務端呢?我想到使用Yii2時，用到redis我是沒有安裝官方提供的redis擴充套件，但是它仍然可以與redis服務端通訊，於是乎便去追蹤了Yii2-redis元件的原始碼，看完以後，深感作者的強大。
準備
看了大佬的程式碼，自己也想去實現一下，於是乎也開始找一些資料。
-
學習redis協議【http://redis.cn/topics/protocol】
1. Redis在TCP埠6379上監聽到來的連線，客戶端連線到來時，Redis伺服器為此建立一個TCP連線。在客戶端與伺服器端之間傳輸的每個Redis命令或者資料都以\r\n結尾 2. 新的統一協議已在Redis 1.2中引入，但是在Redis 2.0中，這就成為了與Redis伺服器通訊的標準方式。 -
理解協議並參照大佬自己去寫
程式碼有很多註釋，看完協議，跟著程式碼就能明白了
<?php /** * Class SocketException */ class SocketException extends \Exception { /** * @var array */ public $errorInfo = []; /** * @return string */ public function getName() { return 'Redis Socket Exception'; } /** * SocketException constructor. * @param string $message * @param array $errorInfo * @param int $code * @param Throwable|null $previous */ public function __construct($message = "", $errorInfo = [], $code = 0, Throwable $previous = null) { $this->errorInfo = $errorInfo; parent::__construct($message, $code, $previous); } /** * @return string */ public function __toString() { return parent::__toString() . PHP_EOL . 'Additional Information:' . PHP_EOL . print_r($this->errorInfo, true);; // TODO: Change the autogenerated stub } } /** * Class RedisConnection */ class RedisConnection { /** * @var string */ public $host = 'localhost'; /** * @var int */ public $port = 6379; /** * @var null */ public $password = null; /** * @var int */ public $database = 0; /** * @var */ public $connectionTimeout; /** * @var null */ public $dataTimeout = null; /** * * STREAM_CLIENT_ASYNC_CONNECT指示應開啟每個後續連線，而不必等待上一個連線的完成 * @var int */ public $socketClientFlags = STREAM_CLIENT_CONNECT; /** * @var array */ protected $pools = []; /** * @var int */ protected $maxPoolSize = 10; /** * RedisConnection constructor. * @param string $host * @param int $port * @param null $password */ public function __construct($host = 'localhost', $port = 6379, $password = null) { $this->host = $host; $this->port = $port; $this->password = $password; } /** * @return string */ public function getConnectionString() { return 'tcp://' . $this->host . ':' . $this->port; } /** * @param int $database */ public function connect($database = 0) { $this->database = $database; $countSize = count($this->pools); if ($countSize > $this->maxPoolSize) { return; } if ($this->getSocket() !== false) { return; } $connId = $this->getConnectionString(); $connection = $connId . ', database=' . $this->database; try { $socket = stream_socket_client( $connId, $errorNumber, $errorDescription, $this->connectionTimeout ?? ini_get('default_socket_timeout'), $this->socketClientFlags ); if ($socket) { $this->pools[$connId] = $socket; if ($this->dataTimeout !== null) { $timeout = (int)$this->dataTimeout; $microTimeout = (int)(($this->dataTimeout - $timeout) * 1000000); stream_set_timeout($socket, $timeout, $microTimeout); } if ($this->password !== null) { $this->exec('AUTH', [$this->password]); } if ($this->database !== null) { $this->exec('SELECT', [$this->database]); } } else { $message = "無法開啟redis資料庫連線 ($connection): $errorNumber - $errorDescription"; throw new Exception($message, $errorDescription, $errorNumber); } } catch (Exception $e) { exit($e->getMessage()); } } /** * 用單行回覆，回覆的第一個位元組將是“+” * 錯誤訊息，回覆的第一個位元組將是“-” * 整型數字，回覆的第一個位元組將是“:” * 批量回復，回覆的第一個位元組將是“$” * 多個批量回復，回覆的第一個位元組將是“*” * 命令LRNGE需要返回多個值 * LRANGE mylist 0 3 * 無值：*0 * 有值: * *3 * $1 * c * $1 * b * $1 * a * sadd mylist a b c d * :4 * * @link http://redis.cn/topics/protocol * @param $cmd * @param array $params */ public function exec($cmd, $params = []) { //狀態：-ERR Client sent AUTH, but no password is set //狀態：+OK //get: $15\r\nlemon1024026382\r\n //del： :1 $params = array_merge(explode(' ', $cmd), $params); $command = ''; $paramsCount = 0; foreach ($params as $param) { if ($param === null) { continue; } $command .= '$' . mb_strlen($param, '8bit') . "\r\n" . $param . "\r\n"; $paramsCount++; } $command = '*' . $paramsCount . "\r\n" . $command; // echo 'Executing Redis Command:', $cmd, PHP_EOL; // echo 'Yuan Shi Redis Cmd:', $command, PHP_EOL; return $this->send($command, $params); } /** * @param $cmd * @param $params * @return null * @throws SocketException */ private function send($cmd, $params) { $socket = $this->getSocket(); $written = fwrite($socket, $cmd); if ($written === false) { throw new SocketException("無法寫入到socket.\nRedis命令是: " . $cmd); } return $this->parseData($params, $cmd); } /** * 用單行回覆，回覆的第一個位元組將是“+” * 錯誤訊息，回覆的第一個位元組將是“-” * 整型數字，回覆的第一個位元組將是“:” * 批量回復，回覆的第一個位元組將是“$” * 多個批量回復，回覆的第一個位元組將是“*” * @link http://redis.cn/topics/protocol * * \r\n作為分割符號，使用fgets按行讀取 * 如果是狀態返回的命令或者錯誤返回或者返回整數這些相對簡單處理 * 批量返回時，如果是*號，第一步先讀取返回的計數器，比如keys * 會返回有多少個key值，迴圈計數器 * ，讀取第二行就是$數字(批量回復)，例如$9表示當前key的value值長度 * 程式碼裡有 $length = intval($line)+2; +2其實是表示\r\n這個分隔符 * @param array $params * @throws SocketException * @return null|string|mixed */ private function parseData($params) { $socket = $this->getSocket(); $prettyCmd = implode(' ', $params); if (($line = fgets($socket)) === false) {// 從檔案指標中讀取一行，這裡最合適，因為協議分割標識是\r\n throw new SocketException("無法從socket讀取.\nRedis命令是: " . $prettyCmd); } echo '服務端響應資料:', $line, PHP_EOL; $type = $line[0]; $line = mb_substr($line, 1, -2, '8bit'); if ('+' === $type) { // eg:SET ping select quit auth... if (in_array($line, ['Ok', 'PONG'])) { return true; } else { return $line; } } if ('-' === $type) { throw new SocketException("Redis 錯誤: " . $line . "\nRedis命令是: " . $prettyCmd); } // eg: hset zadd del if (':' === $type) { return $line; } /** * 例如: * 輸入存在的key:get jie 返回:$9\r\nyangjiecheng\r\n * 輸入:keys * 返回: *9\r\n$9\r\nchat.user\r\n$3\r\njie\r\n$4\r\ntest\r\n * 輸入不存在的key: get 111 返回:$-1 對應redis返回：nil * */ if ('$' === $type) { if (-1 == $line) { return null; } $length = intval($line) + 2; //+2 表示後面的\r\n $data = ''; while ($length > 0) { $str = fread($socket, $length); if ($str === false) { throw new SocketException("無法從socket讀取.\nRedis命令是: " . $prettyCmd); } $data .= $str; $length -= mb_strlen($str, '8bit');//此函式可以指定字串編碼並計算長度，8bit就是按照位元組計算長度 } return !empty($data) ? mb_substr($data, 0, -2, '8bit') : ''; } if ('*' === $type) { $count = intval($line); $data = []; for ($i = 0; $i < $count; $i++) { $data[] = $this->parseData($params); } return $data; } throw new SocketException('從Redis讀取到未能解析的資料: ' . $line . "\nRedis命令是: " . $prettyCmd); } /** * @return bool|mixed */ public function getSocket() { $id = $this->getConnectionString(); return isset($this->pools[$id]) ? $this->pools[$id] : false; } /** * */ public function close() { $connectionString = $this->getConnectionString(); foreach ($this->pools as $socket) { $connection = $connectionString . ', database=' . $this->database; echo 'Closing DB connection: ' . $connection . PHP_EOL; try { $this->exec('QUIT'); } catch (SocketException $e) { // ignore errors when quitting a closed connection } fclose($socket); } $this->pools = []; } /** * */ public function __destruct() { // TODO: Implement __destruct() method. $this->close(); } } /** * @param $file * @param $content */ function writeLog($file, $content) { $str = '@@@@@@@@@@ Time Is ' . date('Y-m-d H:i:s') . ' @@@@@@@@@' . PHP_EOL; $str .= $content . PHP_EOL; $str .= '@@@@@@@@@@ End Block Log @@@@@@@@' . PHP_EOL; file_put_contents($file, $str, FILE_APPEND); } set_exception_handler(/** * @param Exception $exception */ function (\Exception $exception) { echoMsg($exception->getMessage()); }); /** * @param $msg * @param string $type */ function echoMsg($msg, $type = 'error') { if ($type === 'error') { $msg = "\e[" . "1;37m" . $msg . "\e[0m"; } else { $msg = "\e[" . "0;31m" . $msg . "\e[0m"; } echo "\033[" . "41m", "Exception: ", $msg, "\033[0m", PHP_EOL; } set_error_handler(/** * @param $errno * @param $errstr * @param $errfile * @param $errline */ function ($errno, $errstr, $errfile, $errline){ echoMsg("custom error:[$errno] $errstr"); echoMsg(" Error on line $errline in $errfile"); }); //register_shutdown_function $redis = new RedisConnection(); $redis->connect(); echo $redis->getConnectionString(), PHP_EOL; //$redis->exec('SET', ['RedisClient', 'OK']); //$redis->exec('DEL', ['RedisClient']); //$redis->exec('KEYS', ['*']); var_dump($redis->exec('GET', ['access_token'])); //var_dump($redis->exec('LRANGE', ['qlist', 0, 3]));