redis协议相当简单好理解。
redis协议支持类型:
- 正确 Simple Strings
- 错误 Errors
- 整数 Integers
- 字符块 Bulk Strings
- 数组 Arrays
Simple Strings(+)
Simple Strings的呈现以+开始,以\r\n结尾,一个只含有'OK'的表示为+OK\r\n。所以简单的字符串不能包含连续的\r\n字符。
Errors(-)
Errors 的呈现以-开始,以\r\n结尾,一条表示Error message错误信息的字符为-Error message\r\n
Integers(:)
Integers 的呈现以:开始,以\r\n结尾,比如:1000\r\n表示数字1000
Bulk Strings($)
Bulk Strings 数据头同时表示下一行数据长度,不包括换行符长度\r\n,$后面则是对应的长度的数据。它还有一个特殊用途,用长度为-1的空字符表示NUll值
Arrays(*)
Arrays 头以 * 开始,后边接一个Integer类型,表示消息体总共有多少行,不包括当前行,*后面是具体的行数,也就是数组长度,之后接着各自的数据类型。
发送方式
客户端将命令用块字符的方式发送给服务端,比如:
SET xyz abcde
*3\r\n
3ドル\r\n
SET\r\n
3ドル\r\n
xyz\r\n
5ドル\r\n
abcde\r\n
成功或者失败:
+OK\r\n
-错误信息\r\n
如果是内建的命令操作,可以直接发送命令给服务器。比如PING, EXISTS mykey等.使用telnet 127.0.0.1 6379连上默认的redis server, 敲入PING就可以返回结果+PONG
php实现redis客户端和服务端 string功能:
- 服务端
<?php
/**
* 多进程阻塞式
*/
class Xtgxiso_server
{
private $socket = false;
private $process_num = 100;
public $redis_kv_data = array();
public $onMessage = null;
function __construct($host="0.0.0.0",$port=1215)
{
$this->socket = stream_socket_server("tcp://".$host.":".$port,$errno, $errstr);
if (!$this->socket) die($errstr."--".$errno);
echo "listen $host $port \r\n";
ini_set("memory_limit", "128M");
}
private function parseRESP(&$conn){
$line = fgets($conn);
if($line === '' || $line === false)
{
return null;
}
$type = $line[0];
$line = mb_substr($line,1,-2);
switch ( $type ){
case "*":
$count = (int) $line;
$data = array();
for ($i = 1; $i <= $count; $i++) {
$data[] = $this->parseRESP($conn);
}
return $data;
case "$":
if ($line == '-1') {
return null;
}
$length = $line + 2;
$data = '';
while ($length > 0) {
$block = fread($conn, $length);
if ($length !== strlen($block)) {
throw new Exception('RECEIVING');
}
$data .= $block;
$length -= mb_strlen($block);
}
return mb_substr($data, 0, -2);
}
return $line;
}
private function start_worker_process(){
$pid = pcntl_fork();
switch ($pid) {
case -1:
echo "fork error : {$i} \r\n";
exit;
case 0:
while ( 1 ) {
echo "waiting...\n";
$conn = stream_socket_accept($this->socket, -1);
if ( !$conn ){
continue;
}
//"*3\r\n3ドル\r\nSET\r\n5ドル\r\nmykey\r\n7ドル\r\nmyvalue\r\n"
while(1){
$arr = $this->parseRESP($conn);
if ( is_array($arr) ) {
if ($this->onMessage) {
call_user_func($this->onMessage, $conn, $arr);
}
}else if ( $arr ){
if ($this->onMessage) {
call_user_func($this->onMessage, $conn, $arr);
}
}else{
fclose($conn);
break;
}
}
}
default:
$this->pids[$pid] = $pid;
break;
}
}
public function run(){
for($i = 1; $i <= $this->process_num; $i++){
$this->start_worker_process();
}
while(1){
foreach ($this->pids as $i => $pid) {
if($pid) {
$res = pcntl_waitpid($pid, $status,WNOHANG);
if ( $res == -1 || $res > 0 ){
$this->start_worker_process();
unset($this->pids[$pid]);
}
}
}
sleep(1);
}
}
}
$server = new Xtgxiso_server();
$server->onMessage = function($conn,$info) use($server){
if ( is_array($info) ){
if ( $info["0"] == "SET" ) {
$key = $info[1];
$val = $info[2];
$server->redis_kv_data[$key] = $val;
fwrite($conn, "+OK\r\n");
}else if ( $info["0"] == "GET" ){
$key = $info[1];
fwrite($conn, "$".strlen($server->redis_kv_data[$key])."\r\n".$server->redis_kv_data[$key]."\r\n");
}else{
fwrite($conn,"+OK\r\n");
}
}else{
fwrite($conn,"+OK\r\n");
}
};
$server->run();
通过如下命令来测试PHP实现的性能:
redis-benchmark -h 127.0.0.1 -p 1215 -t set -n 80000 -q
- 客户端
<?php
namespace xtgxiso;
class Redis {
private $redis_socket = false;
private $cmd = '';
public function __construct($host='127.0.0.1',$port=6379,$timeout = 3) {
$this->redis_socket = stream_socket_client("tcp://".$host.":".$port, $errno, $errstr, $timeout);
if ( !$this->redis_socket) {
throw new Exception("{$errno} - {$errstr}");
}
}
public function __destruct() {
fclose($this->redis_socket);
}
public function __call($name, $args) {
$crlf = "\r\n";
array_unshift($args,$name);
$command = '*' . count($args) . $crlf;
foreach ($args as $arg) {
$command .= '$' . strlen($arg) . $crlf . $arg . $crlf;
}
$fwrite = fwrite($this->redis_socket,$command);
if ($fwrite === FALSE || $fwrite <= 0) {
throw new Exception('Failed to write entire command to stream');
}
return $this->readResponse();
}
private function readResponse() {
$reply = trim(fgets($this->redis_socket, 1024));
switch (substr($reply, 0, 1)) {
case '-':
throw new Exception(trim(substr($reply, 4)));
break;
case '+':
$response = substr(trim($reply), 1);
if ($response === 'OK') {
$response = TRUE;
}
break;
case '$':
$response = NULL;
if ($reply == '$-1') {
break;
}
$read = 0;
$size = intval(substr($reply, 1));
if ($size > 0) {
do {
$block_size = ($size - $read) > 1024 ? 1024 : ($size - $read);
$r = fread($this->redis_socket, $block_size);
if ($r === FALSE) {
throw new Exception('Failed to read response from stream');
} else {
$read += strlen($r);
$response .= $r;
}
} while ($read < $size);
}
fread($this->redis_socket, 2); /* discard crlf */
break;
/* Multi-bulk reply */
case '*':
$count = intval(substr($reply, 1));
if ($count == '-1') {
return NULL;
}
$response = array();
for ($i = 0; $i < $count; $i++) {
$response[] = $this->readResponse();
}
break;
/* Integer reply */
case ':':
$response = intval(substr(trim($reply), 1));
break;
default:
throw new RedisException("Unknown response: {$reply}");
break;
}
return $response;
}
}
/*
$redis = new Client_test();
var_dump($redis->auth("123456"));
var_dump($redis->set("a",'b'));
var_dump($redis->get("a"));
*/
参考:
http://redis.cn/topics/protocol.html
http://www.01happy.com/php-redis-server/
http://www.01happy.com/php-redis-client/