Redis学习笔记（十）Pipelining

Redis还提供了一种技术——Pipelining（流水线？管线？还是管道？），姑且翻译成管道吧。使用Redis的管道技术可以同时发送多个命令，以节省往返时间。

1.RTT

Redis是一个基于TCP连接的客户/服务器模型，这种模型通常由客户端发起请求，服务端接收请求并作出响应。

对Redis而言，这个过程一般是这样：客户端向服务端发起查询请求，并从套接字中读取服务端的响应数据（一般是以阻塞模式读取）;服务端处理客户端发来的命令，并将命令执行的结果通过套接字返还给客户端。

比如，有如下请求/响应序列：

Client: INCR X
Server: 1
Client: INCR X
Server: 2
Client: INCR X
Server: 3
Client: INCR X
Server: 4

Client: INCR X

Server: 1

Client: INCR X

Server: 2

Client: INCR X

Server: 3

Client: INCR X

Server: 4

需要明确的是，客户端和服务端是通过网络连接通信的。从客户端发送请求包，到客户端再接收到响应包的这段时间，我们称之为往返时间（Round Trip Time, RTT）。所以，当客户端连续的执行多个请求命令时（比如向列表中不断的添加元素），势必会影响到Redis的性能。

举个例子，如果RTT为250ms（已经很慢了），那么服务器每秒最能就只能处理4个请求，即使你服务器有很好的性能也没有用。

2.管道（Pipelining）

Redis使用管道技术，可以实现一次向服务器发送多个命令，并只响应一次，可以极大的节省时间。比如：

Client: INCR X
Client: INCR X
Client: INCR X
Client: INCR X
Server: 1
Server: 2
Server: 3
Server: 4

Client: INCR X

Server: 1

Server: 2

Server: 3

Server: 4

需要注意的是：当客户端使用管道技术向服务器发送多个命令后，服务器会将每个命令的执行结果添加到内存中。因此，在使用管道技术时，要合理的控制批量发送命令的个数。

3.系统调用

你以为管道技术减少的仅仅是RTT消耗的时间成本吗？NO！管道技术不仅仅靠减少RTT来提高性能，更多的是通过减少系统调用来减少的。

我们知道，系统调用意味着用户态与内核态之间的转换，这种转换往往是很耗时的。而每一次的请求/响应都会执行Socket I/O（调用read()和write()系统函数），上下文的切换必然耗费时间。使用管道技术后，我们就可以只执行一次I/O，速度当然会有提升了。

4.示例

使用Redis C客户端，观察Pipelining对速度提升的情况：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <assert.h>
#include "hiredis/hiredis.h"

static long long usec()
{
        struct timeval tv;
        gettimeofday(&tv,NULL);
        return (((long long)tv.tv_sec)*1000000)+tv.tv_usec;
}

int main()
{
        /*连接redis*/
        redisContext *c = redisConnect("127.0.0.1", 6379);
        if(c == NULL){
                printf("Connection error: can't allocate redis context\n");
                exit(1);
        }
        else if(c->err){
                printf("Connection error: %s\n", c->errstr);
                redisFree(c);
                exit(1);
        }
        printf("连接已建立。\n");

        /*不使用Pipelining*/
        redisReply **replies;
        int i, num;
        long long t1, t2;

        num = 1000;
        replies = malloc(sizeof(redisReply*)*num);
        t1 = usec();
        for(i=0; i<num; i++){
                replies[i] = redisCommand(c,"PING");
                assert(replies[i] != NULL && replies[i]->type == REDIS_REPLY_STATUS);
        }
        t2 = usec();
        for(i=0; i<num; i++){
                freeReplyObject(replies[i]);
        }
        free(replies);
        printf("Without Pipelining: %.3fs\n", (t2-t1)/1000000.0);

        /*使用Pipelining*/
        num = 1000;
        replies = malloc(sizeof(redisReply)*num);
        for(i=0; i<num; i++){
                redisAppendCommand(c, "PING");
        }
        t1 = usec();
        for(i=0; i<num; i++){
                assert(redisGetReply(c, (void*)&replies[i]) == REDIS_OK);
                assert(replies[i] != NULL && replies[i]->type == REDIS_REPLY_STATUS);

        }
        t2 = usec();
        for(i=0; i<num; i++){
                freeReplyObject(replies[i]);
        }
        free(replies);
        printf("With Pipelining: %.3fs\n", (t2-t1)/1000000.0);

        redisFree(c);
        return 0;
}

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include <time.h>

#include <assert.h>

#include "hiredis/hiredis.h"

static long long usec()

{

struct timeval tv;

gettimeofday(&tv,NULL);

return (((long long)tv.tv_sec)*1000000)+tv.tv_usec;

}

int main()

{

/*连接redis*/

redisContext *c = redisConnect("127.0.0.1", 6379);

if(c == NULL){

printf("Connection error: can't allocate redis context\n");

exit(1);

}

else if(c->err){

printf("Connection error: %s\n", c->errstr);

redisFree(c);

exit(1);

}

printf("连接已建立。\n");

/*不使用Pipelining*/

redisReply **replies;

int i, num;

long long t1, t2;

num = 1000;

replies = malloc(sizeof(redisReply*)*num);

t1 = usec();

for(i=0; i<num; i++){

replies[i] = redisCommand(c,"PING");

assert(replies[i] != NULL && replies[i]->type == REDIS_REPLY_STATUS);

}

t2 = usec();

for(i=0; i<num; i++){

freeReplyObject(replies[i]);

}

free(replies);

printf("Without Pipelining: %.3fs\n", (t2-t1)/1000000.0);

/*使用Pipelining*/

num = 1000;

replies = malloc(sizeof(redisReply)*num);

for(i=0; i<num; i++){

redisAppendCommand(c, "PING");

}

t1 = usec();

for(i=0; i<num; i++){

assert(redisGetReply(c, (void*)&replies[i]) == REDIS_OK);

assert(replies[i] != NULL && replies[i]->type == REDIS_REPLY_STATUS);

}

t2 = usec();

for(i=0; i<num; i++){

freeReplyObject(replies[i]);

}

free(replies);

printf("With Pipelining: %.3fs\n", (t2-t1)/1000000.0);

redisFree(c);

return 0;

}

编译、运行：

[root@localhost demo]# gcc -Wall -g pipelining.c -o pipelining -lhiredis
[root@localhost demo]# ./pipelining
连接已建立。
Without Pipelining: 0.041s
With Pipelining: 0.003s

[root@localhost demo]# gcc -Wall -g pipelining.c -o pipelining -lhiredis

[root@localhost demo]# ./pipelining

连接已建立。

Without Pipelining: 0.041s

With Pipelining: 0.003s

不使用Pipelining耗时41ms，使用Pipelining耗时3ms。

参考：
https://redis.io/topics/pipelining

Redis学习笔记（十）Pipelining

《Redis学习笔记（十）Pipelining》有1个想法

发表评论取消回复

《Redis学习笔记（十）Pipelining》有1个想法

发表评论 取消回复

发表评论取消回复