在设计分布式系统时，除了考虑并发数及扩展能力外，很多场景下需要考虑相应时间。在供网络带宽和平台确定的情况，保证高并发的同时确保响应时间提高吞吐量，主要有两个方面思路：
- 减少网络交互次数。
- 选择压缩比较高的数据存储协议。例如，存储google buffer的协议的消息而不是文本或者JSON。
对于Redis来说，提高网络使用率降低响应时间，至少可以从以下三个方面考虑：
- 1、使用m系统命令
- 2、使用pipeline
- 3、使用异步代替同步
1、 使用m系统命令
Redis提供m系列的命令可以支持多条命令一下操作。
- 对于string，可以进行mset/mget
- 对于hash，可以进行hgetall，hmget/hmset
如果是普通的规则数据，建模时使用string/hash方式存储，可以考虑使用上述命令进行批量操作提高性能。
2、使用pipeline
Redis 管道技术可以在服务端未响应时，客户端可以继续向服务端发送请求，并最终一次性读取所有服务端的响应。使用redis的C客户端支持pipelie的示例如下：
redisReply *reply;
redisAppendCommand(context,"SET foo bar");
redisAppendCommand(context,"GET foo");
redisGetReply(context,&reply); // reply for SET
freeReplyObject(reply);
redisGetReply(context,&reply); // reply for GET
freeReplyObject(reply);
3、使用异步代替同步
Redis是一种基于客户端-服务端模型以及请求/响应协议的TCP服务。这意味着通常情况下一个请求会遵循以下步骤：
- 客户端向服务端发送一个查询请求，并监听Socket返回，通常是以阻塞模式，等待服务端响应。
- 服务端处理命令，并将结果返回给客户端。
这是一种典型的同步模式，接口如下。
redisContext *redisConnect(const char *ip, int port);
void *redisCommand(redisContext *c, const char *format, ...);
void freeReplyObject(void *reply);
异步和同步不同的是，发生命令时，无需等待返回，可以继续发送下一个命令，从而提高了性能。
redisAsyncContext *redisAsyncConnect(const char *ip, int port);
int redisAsyncSetConnectCallback(redisAsyncContext *ac, redisConnectCallback *fn);
int redisAsyncSetDisconnectCallback(redisAsyncContext *ac, redisDisconnectCallback *fn);
void redisAsyncDisconnect(redisAsyncContext *ac);
void redisAsyncFree(redisAsyncContext *ac);
/* Command functions for an async context. Write the command to the
* output buffer and register the provided callback. */
int redisvAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, va_list ap);
int redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, ...);
异步事件模型可以是ae，libevent，libuv，libev等。使用ae示例代码如下：
int main (int argc, char **argv) {
signal(SIGPIPE, SIG_IGN);
redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);
if (c->err) {
/* Let *c leak for now... */
printf("Error: %s\n", c->errstr);
return 1;
}
loop = aeCreateEventLoop(64);
redisAeAttach(loop, c);
redisAsyncSetConnectCallback(c,connectCallback);
redisAsyncSetDisconnectCallback(c,disconnectCallback);
redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1]));
redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key");
aeMain(loop);
return 0;
}
注意：最后一次redisAsyncCommand时，在回调函数getCallback中需要redisAsyncDisconnect。
小结
综合上述三种方式，第一种最简单，第三种方式比较复杂，需要通过回掉函数进行相关额外操作。最优选择1，如果不满足时，可以不满足时，需要考虑结合第二种方式。在Cluster模式需要考虑Batch的操作最好在一个Node，否则会存在跳转，影响性能。另外每个batch的数据量不宜太大，否则会造成拥塞影响后续操作的影响时间。
