pipeline 压缩请求数量

通常情况下，我们每个操作 Redis 的命令都以一个 TCP 请求发送给 Redis，这样的做法简单直观。然而，当我们有连续多个命令需要发送给 Redis 时，如果每个命令都以一个数据包发送给 Redis，将会降低服务端的并发能力。

为什么呢？大家知道每发送一个 TCP 报文，会存在网络延时及操作系统的处理延时。大部分情况下，网络延时要远大于 CPU 的处理延时。如果一个简单的命令就以一个 TCP 报文发出，网络延时将成为系统性能瓶颈，使得服务端的并发数量上不去。

首先检查你的代码，是否明确完整使用了 Redis 的长连接机制。作为一个服务端程序员，要对长连接的使用有一定了解，在条件允许的情况下，一定要开启长连接。验证方式也比较简单，直接用 tcpdump 或 wireshark 抓包分析一下网络数据即可。

set_keepalive的参数：按照业务正常运转的并发数量设置，不建议使用峰值情况设置。

如果我们确定开启了长连接，发现这时候 Redis 的 CPU 的占用率还是不高，在这种情况下，就要从 Redis 的使用方法上进行优化。

如果我们可以把所有单次请求，压缩到一起，如下图：

请求示意图

很庆幸 Redis 早就为我们准备好了这道菜，就等着我们吃了，这道菜就叫 pipeline。pipeline 机制将多个命令汇聚到一个请求中，可以有效减少请求数量，减少网络延时。下面是对比使用 pipeline 的一个例子：

  1. # you do not need the following line if you are using
  2.     # the ngx_openresty bundle:
  3.     lua_package_path "/path/to/lua-resty-redis/lib/?.lua;;";
  5.     server {
  6.         location /withoutpipeline {
  7.            content_by_lua_block {
  8.                 local redis = require("resty.redis")
  9.                 local red = redis:new()
  11.                 red:set_timeout(1000) -- 1 sec
  13.                 -- or connect to a unix domain socket file listened
  14.                 -- by a redis server:
  15.                 --     local ok, err = red:connect("unix:/path/to/redis.sock")
  17.                 local ok, err = red:connect("127.0.0.1", 6379)
  18.                 if not ok then
  19.                     ngx.say("failed to connect: ", err)
  20.                     return
  21.                 end
  23.                 local ok, err = red:set("cat", "Marry")
  24.                 ngx.say("set result: ", ok)
  25.                 local res, err = red:get("cat")
  26.                 ngx.say("cat: ", res)
  28.                 ok, err = red:set("horse", "Bob")
  29.                 ngx.say("set result: ", ok)
  30.                 res, err = red:get("horse")
  31.                 ngx.say("horse: ", res)
  34.                 -- put it into the connection pool of size 100,
  35.                 -- with 10 seconds max idle time
  36.                 local ok, err = red:set_keepalive(10000, 100)
  37.                 if not ok then
  38.                     ngx.say("failed to set keepalive: ", err)
  39.                     return
  40.                 end
  41.             }
  42.         }
  44.         location /withpipeline {
  45.             content_by_lua_block {
  46.                 local redis = require("resty.redis")
  47.                 local red = redis:new()
  49.                 red:set_timeout(1000) -- 1 sec
  51.                 -- or connect to a unix domain socket file listened
  52.                 -- by a redis server:
  53.                 --     local ok, err = red:connect("unix:/path/to/redis.sock")
  55.                 local ok, err = red:connect("127.0.0.1", 6379)
  56.                 if not ok then
  57.                     ngx.say("failed to connect: ", err)
  58.                     return
  59.                 end
  61.                 red:init_pipeline()
  62.                 red:set("cat", "Marry")
  63.                 red:set("horse", "Bob")
  64.                 red:get("cat")
  65.                 red:get("horse")
  66.                 local results, err = red:commit_pipeline()
  67.                 if not results then
  68.                     ngx.say("failed to commit the pipelined requests: ", err)
  69.                     return
  70.                 end
  72.                 for i, res in ipairs(results) do
  73.                     if type(res) == "table" then
  74.                         if not res[1] then
  75.                             ngx.say("failed to run command ", i, ": ", res[2])
  76.                         else
  77.                             -- process the table value
  78.                         end
  79.                     else
  80.                         -- process the scalar value
  81.                     end
  82.                 end
  84.                 -- put it into the connection pool of size 100,
  85.                 -- with 10 seconds max idle time
  86.                 local ok, err = red:set_keepalive(10000, 100)
  87.                 if not ok then
  88.                     ngx.say("failed to set keepalive: ", err)
  89.                     return
  90.                 end
  91.             }
  92.         }
  93.     }

在我们实际应用场景中，正确使用 pipeline 对性能的提升十分明显。我们曾经某个后台应用，逐个处理大约 100 万条记录需要几十分钟，经过 pileline 压缩请求数量后，最后时间缩小到 20 秒左右。做之前能预计提升性能，但是没想到提升如此巨大。