使用原生PHP编写Redis扩展（客户端）介绍，phpredis

使用原生PHP编写Redis扩展（客户端）介绍，phpredis

Redis是典型的C/S架构软件，Client 和 Server 之间通过TCP连接进行通信，所以原则上只要是支持socket编程的语言都可以用来编写Redis的客户端，PHP自然也不例外，只是一般我们习惯上或出于性能考虑，使用C编写的Redis扩展。

这里可以使用简单易用的 stream_socket_*  族函数来进行socket编程

比如连接Redis服务器（假设为127.0.0.1:6379），可以使用以下代码：

<?php

$redis = stream_socket_client('tcp://127.0.0.1:6379', $errno, $errstr, 5);

if (!$redis)
{
    die('连接redis服务器失败: ' . $errstr);
}

// 查询代码....

stream_socket_shutdown($redis, STREAM_SHUT_RDWR);

要编写Redis扩展，首先我们得了解Redis客户端和服务端之间的通信协议，官方称之为 RESP（REdis Serialization Protocol），这个协议其实还是很简单易懂的，下面我们简单介绍下：

1、所有命令和数据以 "\r\n" 结尾

2、服务器根据执行的命令返回不同类型的结果，不同的数据类型用第一个字符标识，具体如下：

• "+" 服务器返回一个简单字符串结果 比如 set foo bar 命令返回 +OK\r\n
• "-"  命令执行出错，比如  -WRONGTYPE Operation against a key holding the wrong kind of value\r\n
•  ":"  整数结果，比如 dbsize 命令返回  :1000\r\n
• "$"  二进制安全的长字符串结果，比如 get foo 命令返回  $3\r\nbar\r\n  其中数字3的位置表示字符串长度，字符串被一对\r\n包含
• "*"  返回结果是一个数组，比如 hkeys foobar 命令返回  *2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n  其中*后数字2的位置表示元素个数，之后可以是以上各种基本类型的组合。

3、客户端发送命令通过长字符串数组形式，例如要查询 set foobar hello 命令写成  *3\r\n$3\r\nset\r\n$6\r\nfoobar\r\n$5\r\nhello\r\n

更详细的协议内容可以查看官方文档   https://redis.io/topics/protocol

接下来用代码简单演示：

<?php

$redis = stream_socket_client('tcp://127.0.0.1:6379', $errno, $errstr, 5);

if (!$redis)
<?php

$redis = stream_socket_client('tcp://127.0.0.1:6379', $errno, $errstr, 5);

if (!$redis)
{
    die('连接redis服务器失败: ' . $errstr);
}


// 查询代码....
$cmd = "*1\r\n$6\r\nDBSIZE\r\n";  //dbsize
fwrite($redis, $cmd, strlen($cmd));
$ret = fread($redis, 4096);
echo $ret;
echo "----------------------\r\n";

$cmd = "*3\r\n$3\r\nset\r\n$6\r\nfoobar\r\n$5\r\nredis\r\n"; //set foobar redis
fwrite($redis, $cmd, strlen($cmd));
$ret = fread($redis, 4096);
echo $ret;
echo "----------------------------\r\n";

$cmd = "*2\r\n$3\r\nget\r\n$6\r\nfoobar\r\n";  //get foobar
fwrite($redis, $cmd, strlen($cmd));
$ret = fread($redis, 4096);
echo $ret;
echo "----------------------------\r\n";

$cmd = "*4\r\n$4\r\nhset\r\n$7\r\nanimals\r\n$3\r\ncat\r\n$3\r\ntom\r\n";  //hset animals cat tom
fwrite($redis, $cmd, strlen($cmd));
$ret = fread($redis, 4096);
echo $ret;
echo "-------------------------------------\r\n";

$cmd = "*2\r\n$5\r\nhkeys\r\n$7\r\nanimals\r\n";  //hkeys animals
fwrite($redis, $cmd, strlen($cmd));
$ret = fread($redis, 4096);
echo $ret;
echo "-------------------------------------\r\n";

stream_socket_shutdown($redis, STREAM_SHUT_RDWR);

执行结果：

[root@localhost php]# php redis.php 
:15
----------------------
+OK
----------------------------
$5
redis
----------------------------
:0
-------------------------------------
*1
$3
cat
-------------------------------------

返回的 \r\n 这里显示为换行

把代码优化封装一下，得到最终代码：

<?php

class PhpRedisException extends Exception{}


class PhpRedis
{
    protected $conn = NULL;

    protected $command = NULL;

    protected $isPipeline = FALSE;

    protected $pipelineCmd = '';

    protected $pipelineCount = 0;

    protected $response = '';

    public function connect($host = '127.0.0.1', $port = 6379, $timeout = 0)
    {
            $this->conn = stream_socket_client("tcp://$host:$port", $errno, $errstr, $timeout);
            if (!$this->conn)
            {
                throw new PhpRedisException("无法连接redis服务器：$errstr", $errno);
            }
    }

    protected function _makeCommand($args)
    {
            $cmds = array();
            $cmds[] = '*' . count($args) . "\r\n";
            foreach($args as $arg)
            {
                $cmds[] = '$' . strlen($arg) . "\r\n$arg\r\n";
            }

            $this->command = implode($cmds);
    }

    protected function _fmtResult()
    {
            if ($this->response[0] == '-')
            {
                $this->response = ltrim($this->response, '-');
                list($errstr, $this->response) = explode("\r\n", $this->response, 2);
                throw new PhpRedisException($errstr, 500);
            }

            switch($this->response[0])
            {
            case '+':
            case ':':
                    list($ret, $this->response) = explode("\r\n", $this->response, 2);
                    $ret = substr($ret, 1);
                    break;
            case '$':
                    $this->response = ltrim($this->response, '$');
                    list($slen, $this->response) = explode("\r\n", $this->response, 2);
                    $ret = substr($this->response, 0, intval($slen));
                    $this->response = substr($this->response, 2 + $slen);
                    break;
            case '*':
                    $ret = $this->_resToArray();
                    break;
            }

            return $ret;
    }

    protected function _resToArray()
    {
            $ret = array();
            $this->response = ltrim($this->response, '*');
            list($count, $this->response) = explode("\r\n", $this->response, 2);
            for($i = 0; $i < $count; $i++)
            {
                    $tmp = $this->_fmtResult();
                    $ret[] = $tmp;
            }
            return $ret;
    }

    protected function _fetchResponse()
    {

            $this->response = fread($this->conn, 8196);
            stream_set_blocking($this->conn, 0); // 设置连接为非阻塞
            // 继续读取返回结果
            while($buf = fread($this->conn, 8196))
            {
                $this->response .= $buf;
            }
            stream_set_blocking($this->conn, 1); // 恢复连接为阻塞
    }

    public function exec()
    {
            if (func_num_args() == 0)
            {
                throw new PhpRedisException("参数不可以为空", 301);
            }
            $this->_makeCommand(func_get_args());

            if (TRUE === $this->isPipeline)
            {
                $this->pipelineCmd .= $this->command;
                $this->pipelineCount++;
                return;
            }

            //echo $this->command;
            fwrite($this->conn, $this->command, strlen($this->command));

            $this->_fetchResponse();

            //echo $this->response;
            return $this->_fmtResult();
    }

    public function initPipeline()
    {
        $this->isPipeline = TRUE;
        $this->pipelineCount = 0;
        $this->pipelineCmd = '';
    }

    public function commitPipeline()
    {
        $ret = array();

        if ($this->pipelineCmd)
        {
            fwrite($this->conn, $this->pipelineCmd, strlen($this->pipelineCmd));

            $this->_fetchResponse();

            for($i = 0; $i < $this->pipelineCount; $i++)
            {
                $ret[] = $this->_fmtResult();
            }
        }
        $this->isPipeline = FALSE;
        $this->pipelineCmd = '';
        return $ret;
    }

    public function close()
    {
        @stream_socket_shutdown($this->conn, STREAM_SHUT_RDWR);
        @fclose($this->conn);
        $this->conn = NULL;
    }
}

调用：

$redis = new PhpRedis();
$redis->connect('127.0.0.1', 6379);

$redis->exec('set', 'foo', 'phpredis');
$redis->exec('hset', 'animals', 'dog', 'spike');
$redis->exec('hset', 'animals', 'cat', 'tom');
$redis->exec('hset', 'animals', 'mouse', 'jerry');

var_dump($redis->exec('get', 'foo'));
var_dump($redis->exec('dbsize'));
print_r($redis->exec('hkeys', 'animals'));

// pipeline
$redis->initPipeline();
$redis->exec('incr', 'Count');
$redis->exec('incr', 'Count');
$redis->exec('hgetall', 'animals');
print_r($redis->commitPipeline());

$redis->close();

运行结果：

string(8) "phpredis"
string(1) "6"
Array
(
    [0] => dog
    [1] => cat
    [2] => mouse
)
Array
(
    [0] => 13
    [1] => 14
    [2] => Array
        (
            [0] => dog
            [1] => spike
            [2] => cat
            [3] => tom
            [4] => mouse
            [5] => jerry
        )


)

That's it^^