Redis 通訊協(xié)議（RESP）

RESP 協(xié)議

微信公眾號(hào)：運(yùn)維開(kāi)發(fā)故事，作者：老鄭

Redis 基于 RESP （Redis Serialization Protocal）協(xié)議來(lái)完成客戶(hù)端和服務(wù)端通訊的。RESP 本質(zhì)是一種文本協(xié)議，實(shí)現(xiàn)簡(jiǎn)單、易于解析。如下表所示：

如果客戶(hù)端和服務(wù)端在一臺(tái)機(jī)器上。那么會(huì)對(duì)通訊協(xié)議進(jìn)行優(yōu)化，直接走本地回環(huán)

我們可以通過(guò) tcpdump 命令來(lái)抓取客戶(hù)端和服務(wù)端請(qǐng)求、響應(yīng)的數(shù)據(jù)包， 命令如下：

# linuxtcpdump -i lo part 6379 -Ann# mac tcpdump -i lo0 port 6379 -Ann

我們以一條 `set msg100 1` 這條命令測(cè)試一下 ( 我本機(jī)是 mac 環(huán)境)：

# 客戶(hù)端 A127.0.0.1:6379> set msg100 1OK

服務(wù)端抓包結(jié)果如下所示：

?  ~ sudo tcpdump -i lo0 port 6379 -AnnPassword:tcpdump: verbose output suppressed, use -v or -vv for full protocol decodelistening on lo0, link-type NULL (BSD loopback), capture size 262144 bytes21:52:53.447885 IP 127.0.0.1.51645 > 127.0.0.1.6379: Flags [P.], seq 1564111974:1564112006, ack 169183468, win 6272, options [nop,nop,TS val 774447713 ecr 772455554], length 32: RESP "set" "msg100" "1"E..T..@.@...............]:tf........H......)"a...*3$3set$6msg100$1121:52:53.447912 IP 127.0.0.1.6379 > 127.0.0.1.51645: Flags [.], ack 32, win 6376, options [nop,nop,TS val 774447713 ecr 774447713], length 0E..4..@.@..................]:t......(......)"a.)"a21:52:53.528935 IP 127.0.0.1.6379 > 127.0.0.1.51645: Flags [P.], seq 1:6, ack 32, win 6376, options [nop,nop,TS val 774447793 ecr 774447713], length 5: RESP "OK"E..9..@.@..................]:t......-......)"..)"a+OK21:52:53.528966 IP 127.0.0.1.51645 > 127.0.0.1.6379: Flags [.], ack 6, win 6272, options [nop,nop,TS val 774447793 ecr 774447793], length 0E..4..@.@...............]:t.........(......)"..)".

redis-cli 客戶(hù)端效果：

客戶(hù)端是對(duì)顯示結(jié)果做了轉(zhuǎn)化，在 redis-cli.c 文件中下面是它的部分源碼

static sds cliFormatReplyTTY(redisReply *r, char *prefix) {    sds out = sdsempty();    switch (r->type) {    case REDIS_REPLY_ERROR:        out = sdscatprintf(out,"(error) %s\n", r->str);    break;    case REDIS_REPLY_STATUS:        out = sdscat(out,r->str);        out = sdscat(out,"\n");    break;    case REDIS_REPLY_INTEGER:        out = sdscatprintf(out,"(integer) %lld\n",r->integer);    break;    case REDIS_REPLY_DOUBLE:        out = sdscatprintf(out,"(double) %s\n",r->str);    break;    case REDIS_REPLY_STRING:    case REDIS_REPLY_VERB:        /* If you are producing output for the standard output we want        * a more interesting output with quoted characters and so forth,        * unless it's a verbatim string type. */        if (r->type == REDIS_REPLY_STRING) {            out = sdscatrepr(out,r->str,r->len);            out = sdscat(out,"\n");        } else {            out = sdscatlen(out,r->str,r->len);            out = sdscat(out,"\n");        }    break;    case REDIS_REPLY_NIL:        out = sdscat(out,"(nil)\n");    break;    case REDIS_REPLY_BOOL:        out = sdscat(out,r->integer ? "(true)\n" : "(false)\n");    break;    case REDIS_REPLY_ARRAY:    case REDIS_REPLY_MAP:    case REDIS_REPLY_SET:    case REDIS_REPLY_PUSH:        if (r->elements == 0) {            if (r->type == REDIS_REPLY_ARRAY)                out = sdscat(out,"(empty array)\n");            else if (r->type == REDIS_REPLY_MAP)                out = sdscat(out,"(empty hash)\n");            else if (r->type == REDIS_REPLY_SET)                out = sdscat(out,"(empty set)\n");            else if (r->type == REDIS_REPLY_PUSH)                out = sdscat(out,"(empty push)\n");            else                out = sdscat(out,"(empty aggregate type)\n");        } else {            unsigned int i, idxlen = 0;            char _prefixlen[16];            char _prefixfmt[16];            sds _prefix;            sds tmp;            /* Calculate chars needed to represent the largest index */            i = r->elements;            if (r->type == REDIS_REPLY_MAP) i /= 2;            do {                idxlen++;                i /= 10;            } while(i);            /* Prefix for nested multi bulks should grow with idxlen+2 spaces */            memset(_prefixlen,' ',idxlen+2);            _prefixlen[idxlen+2] = '\0';            _prefix = sdscat(sdsnew(prefix),_prefixlen);            /* Setup prefix format for every entry */            char numsep;            if (r->type == REDIS_REPLY_SET) numsep = '~';            else if (r->type == REDIS_REPLY_MAP) numsep = '#';            else numsep = ')';            snprintf(_prefixfmt,sizeof(_prefixfmt),"%%s%%%ud%c ",idxlen,numsep);            for (i = 0; i < r->elements; i++) {                unsigned int human_idx = (r->type == REDIS_REPLY_MAP) ?                                         i/2 : i;                human_idx++; /* Make it 1-based. */                /* Don't use the prefix for the first element, as the parent                 * caller already prepended the index number. */                out = sdscatprintf(out,_prefixfmt,i == 0 ? "" : prefix,human_idx);                /* Format the multi bulk entry */                tmp = cliFormatReplyTTY(r->element[i],_prefix);                out = sdscatlen(out,tmp,sdslen(tmp));                sdsfree(tmp);                /* For maps, format the value as well. */                if (r->type == REDIS_REPLY_MAP) {                    i++;                    sdsrange(out,0,-2);                    out = sdscat(out," => ");                    tmp = cliFormatReplyTTY(r->element[i],_prefix);                    out = sdscatlen(out,tmp,sdslen(tmp));                    sdsfree(tmp);                }            }            sdsfree(_prefix);        }    break;    default:        fprintf(stderr,"Unknown reply type: %d\n", r->type);        exit(1);    }    return out;}

我們也可以使用 nc 命令來(lái)替代 redis-cli 命令行：

?  ~ sudo nc 127.0.0.1 6379set a a+OKget a$1a

錯(cuò)誤代碼

• Redis 常見(jiàn)的錯(cuò)誤代碼定義如下：

#define REDIS_ERR -1#define REDIS_OK 0/* When an error occurs, the err flag in a context is set to hold the type of * error that occurred. REDIS_ERR_IO means there was an I/O error and you * should use the "errno" variable to find out what is wrong. * For other values, the "errstr" field will hold a description. */#define REDIS_ERR_IO 1 /* Error in read or write */#define REDIS_ERR_EOF 3 /* End of file */#define REDIS_ERR_PROTOCOL 4 /* Protocol error */#define REDIS_ERR_OOM 5 /* Out of memory */#define REDIS_ERR_TIMEOUT 6 /* Timed out */#define REDIS_ERR_OTHER 2 /* Everything else... */#define REDIS_REPLY_STRING 1#define REDIS_REPLY_ARRAY 2#define REDIS_REPLY_INTEGER 3#define REDIS_REPLY_NIL 4#define REDIS_REPLY_STATUS 5#define REDIS_REPLY_ERROR 6#define REDIS_REPLY_DOUBLE 7#define REDIS_REPLY_BOOL 8#define REDIS_REPLY_MAP 9#define REDIS_REPLY_SET 10#define REDIS_REPLY_ATTR 11#define REDIS_REPLY_PUSH 12#define REDIS_REPLY_BIGNUM 13#define REDIS_REPLY_VERB 14/* Default max unused reader buffer. */#define REDIS_READER_MAX_BUF (1024*16)/* Default multi-bulk element limit */#define REDIS_READER_MAX_ARRAY_ELEMENTS ((1LL<<32) - 1)

• 字符串錯(cuò)誤信息 - 共享對(duì)象

void createSharedObjects(void) {    int j;    /* Shared command responses */    shared.crlf = createObject(OBJ_STRING,sdsnew("\r\n"));    shared.ok = createObject(OBJ_STRING,sdsnew("+OK\r\n"));    shared.emptybulk = createObject(OBJ_STRING,sdsnew("$0\r\n\r\n"));    shared.czero = createObject(OBJ_STRING,sdsnew(":0\r\n"));    shared.cone = createObject(OBJ_STRING,sdsnew(":1\r\n"));    shared.emptyarray = createObject(OBJ_STRING,sdsnew("*0\r\n"));    shared.pong = createObject(OBJ_STRING,sdsnew("+PONG\r\n"));    shared.queued = createObject(OBJ_STRING,sdsnew("+QUEUED\r\n"));    shared.emptyscan = createObject(OBJ_STRING,sdsnew("*2\r\n$1\r\n0\r\n*0\r\n"));    shared.space = createObject(OBJ_STRING,sdsnew(" "));    shared.colon = createObject(OBJ_STRING,sdsnew(":"));    shared.plus = createObject(OBJ_STRING,sdsnew("+"));    /* Shared command error responses */    shared.wrongtypeerr = createObject(OBJ_STRING,sdsnew(        "-WRONGTYPE Operation against a key holding the wrong kind of value\r\n"));    shared.err = createObject(OBJ_STRING,sdsnew("-ERR\r\n"));    shared.nokeyerr = createObject(OBJ_STRING,sdsnew(        "-ERR no such key\r\n"));    shared.syntaxerr = createObject(OBJ_STRING,sdsnew(        "-ERR syntax error\r\n"));    shared.sameobjecterr = createObject(OBJ_STRING,sdsnew(        "-ERR source and destination objects are the same\r\n"));    shared.outofrangeerr = createObject(OBJ_STRING,sdsnew(        "-ERR index out of range\r\n"));    shared.noscripterr = createObject(OBJ_STRING,sdsnew(        "-NOSCRIPT No matching script. Please use EVAL.\r\n"));    shared.loadingerr = createObject(OBJ_STRING,sdsnew(        "-LOADING Redis is loading the dataset in memory\r\n"));    shared.slowscripterr = createObject(OBJ_STRING,sdsnew(        "-BUSY Redis is busy running a script. You can only call SCRIPT KILL or SHUTDOWN NOSAVE.\r\n"));    shared.masterdownerr = createObject(OBJ_STRING,sdsnew(        "-MASTERDOWN Link with MASTER is down and replica-serve-stale-data is set to 'no'.\r\n"));    shared.bgsaveerr = createObject(OBJ_STRING,sdsnew(        "-MISCONF Redis is configured to save RDB snapshots, but it is currently not able to persist on disk. Commands that may modify the data set are disabled, because this instance is configured to report errors during writes if RDB snapshotting fails (stop-writes-on-bgsave-error option). Please check the Redis logs for details about the RDB error.\r\n"));    shared.roslaveerr = createObject(OBJ_STRING,sdsnew(        "-READONLY You can't write against a read only replica.\r\n"));    shared.noautherr = createObject(OBJ_STRING,sdsnew(        "-NOAUTH Authentication required.\r\n"));    shared.oomerr = createObject(OBJ_STRING,sdsnew(        "-OOM command not allowed when used memory > 'maxmemory'.\r\n"));    shared.execaborterr = createObject(OBJ_STRING,sdsnew(        "-EXECABORT Transaction discarded because of previous errors.\r\n"));    shared.noreplicaserr = createObject(OBJ_STRING,sdsnew(        "-NOREPLICAS Not enough good replicas to write.\r\n"));    shared.busykeyerr = createObject(OBJ_STRING,sdsnew(        "-BUSYKEY Target key name already exists.\r\n"));    /* The shared NULL depends on the protocol version. */    shared.null[0] = NULL;    shared.null[1] = NULL;    shared.null[2] = createObject(OBJ_STRING,sdsnew("$-1\r\n"));    shared.null[3] = createObject(OBJ_STRING,sdsnew("_\r\n"));    shared.nullarray[0] = NULL;    shared.nullarray[1] = NULL;    shared.nullarray[2] = createObject(OBJ_STRING,sdsnew("*-1\r\n"));    shared.nullarray[3] = createObject(OBJ_STRING,sdsnew("_\r\n"));    shared.emptymap[0] = NULL;    shared.emptymap[1] = NULL;    shared.emptymap[2] = createObject(OBJ_STRING,sdsnew("*0\r\n"));    shared.emptymap[3] = createObject(OBJ_STRING,sdsnew("%0\r\n"));    shared.emptyset[0] = NULL;    shared.emptyset[1] = NULL;    shared.emptyset[2] = createObject(OBJ_STRING,sdsnew("*0\r\n"));    shared.emptyset[3] = createObject(OBJ_STRING,sdsnew("~0\r\n"));    for (j = 0; j < PROTO_SHARED_SELECT_CMDS; j++) {        char dictid_str[64];        int dictid_len;        dictid_len = ll2string(dictid_str,sizeof(dictid_str),j);        shared.select[j] = createObject(OBJ_STRING,            sdscatprintf(sdsempty(),                "*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n",                dictid_len, dictid_str));    }    shared.messagebulk = createStringObject("$7\r\nmessage\r\n",13);    shared.pmessagebulk = createStringObject("$8\r\npmessage\r\n",14);    shared.subscribebulk = createStringObject("$9\r\nsubscribe\r\n",15);    shared.unsubscribebulk = createStringObject("$11\r\nunsubscribe\r\n",18);    shared.psubscribebulk = createStringObject("$10\r\npsubscribe\r\n",17);    shared.punsubscribebulk = createStringObject("$12\r\npunsubscribe\r\n",19);    /* Shared command names */    shared.del = createStringObject("DEL",3);    shared.unlink = createStringObject("UNLINK",6);    shared.rpop = createStringObject("RPOP",4);    shared.lpop = createStringObject("LPOP",4);    shared.lpush = createStringObject("LPUSH",5);    shared.rpoplpush = createStringObject("RPOPLPUSH",9);    shared.lmove = createStringObject("LMOVE",5);    shared.blmove = createStringObject("BLMOVE",6);    shared.zpopmin = createStringObject("ZPOPMIN",7);    shared.zpopmax = createStringObject("ZPOPMAX",7);    shared.multi = createStringObject("MULTI",5);    shared.exec = createStringObject("EXEC",4);    shared.hset = createStringObject("HSET",4);    shared.srem = createStringObject("SREM",4);    shared.xgroup = createStringObject("XGROUP",6);    shared.xclaim = createStringObject("XCLAIM",6);    shared.script = createStringObject("SCRIPT",6);    shared.replconf = createStringObject("REPLCONF",8);    shared.pexpireat = createStringObject("PEXPIREAT",9);    shared.pexpire = createStringObject("PEXPIRE",7);    shared.persist = createStringObject("PERSIST",7);    shared.set = createStringObject("SET",3);    shared.eval = createStringObject("EVAL",4);    /* Shared command argument */    shared.left = createStringObject("left",4);    shared.right = createStringObject("right",5);    shared.pxat = createStringObject("PXAT", 4);    shared.px = createStringObject("PX",2);    shared.time = createStringObject("TIME",4);    shared.retrycount = createStringObject("RETRYCOUNT",10);    shared.force = createStringObject("FORCE",5);    shared.justid = createStringObject("JUSTID",6);    shared.lastid = createStringObject("LASTID",6);    shared.default_username = createStringObject("default",7);    shared.ping = createStringObject("ping",4);    shared.setid = createStringObject("SETID",5);    shared.keepttl = createStringObject("KEEPTTL",7);    shared.load = createStringObject("LOAD",4);    shared.createconsumer = createStringObject("CREATECONSUMER",14);    shared.getack = createStringObject("GETACK",6);    shared.special_asterick = createStringObject("*",1);    shared.special_equals = createStringObject("=",1);    shared.redacted = makeObjectShared(createStringObject("(redacted)",10));    for (j = 0; j < OBJ_SHARED_INTEGERS; j++) {        shared.integers[j] =            makeObjectShared(createObject(OBJ_STRING,(void*)(long)j));        shared.integers[j]->encoding = OBJ_ENCODING_INT;    }    for (j = 0; j < OBJ_SHARED_BULKHDR_LEN; j++) {        shared.mbulkhdr[j] = createObject(OBJ_STRING,            sdscatprintf(sdsempty(),"*%d\r\n",j));        shared.bulkhdr[j] = createObject(OBJ_STRING,            sdscatprintf(sdsempty(),"$%d\r\n",j));    }    /* The following two shared objects, minstring and maxstrings, are not     * actually used for their value but as a special object meaning     * respectively the minimum possible string and the maximum possible     * string in string comparisons for the ZRANGEBYLEX command. */    shared.minstring = sdsnew("minstring");    shared.maxstring = sdsnew("maxstring");}

命令對(duì)象

redis 命令是使用的是 redisCommand 數(shù)據(jù)結(jié)構(gòu)來(lái)管理的。

typedef void redisCommandProc(client *c);// 函數(shù)指針類(lèi)型，指向命令實(shí)現(xiàn)函數(shù)typedef int redisGetKeysProc(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);struct redisCommand {    char *name;    redisCommandProc *proc;    // 限制命令的個(gè)數(shù)。-N 表示至少 N 個(gè)參數(shù)，包含命令本身    int arity;    // 字符串方式設(shè)置命令的屬性之間運(yùn)用 ｜ 運(yùn)算，程序內(nèi)部自動(dòng)解析，函數(shù) populateCommandTable    char *sflags;   /* Flags as string representation, one char per flag. */    // 將 flags 字符串類(lèi)型轉(zhuǎn)換成整數(shù)，多個(gè)屬性    uint64_t flags; /* The actual flags, obtained from the 'sflags' field. */    /* Use a function to determine keys arguments in a command line.     * Used for Redis Cluster redirect. */    redisGetKeysProc *getkeys_proc;    /* What keys should be loaded in background when calling this command? */    int firstkey; /* The first argument that's a key (0 = no keys) */    int lastkey;  /* The last argument that's a key */    int keystep;  /* The step between first and last key */    long long microseconds, calls, rejected_calls, failed_calls;    int id;     /* Command ID. This is a progressive ID starting from 0 that                   is assigned at runtime, and is used in order to check                   ACLs. A connection is able to execute a given command if                   the user associated to the connection has this command                   bit set in the bitmap of allowed commands. */};

針對(duì) sflag 標(biāo)示，這里可以看看 《redis 設(shè)計(jì)與實(shí)現(xiàn)》

flag 記錄的是 flag 值與 sflag 進(jìn)行運(yùn)算的結(jié)果，見(jiàn) populateCommandTable 函數(shù)

    for (int j = 0; j < argc; j++) {        char *flag = argv[j];        if (!strcasecmp(flag,"write")) {            c->flags |= CMD_WRITE|CMD_CATEGORY_WRITE;        } else if (!strcasecmp(flag,"read-only")) {            c->flags |= CMD_READONLY|CMD_CATEGORY_READ;        } else if (!strcasecmp(flag,"use-memory")) {            c->flags |= CMD_DENYOOM;        } else if (!strcasecmp(flag,"admin")) {            c->flags |= CMD_ADMIN|CMD_CATEGORY_ADMIN|CMD_CATEGORY_DANGEROUS;        } else if (!strcasecmp(flag,"pub-sub")) {            c->flags |= CMD_PUBSUB|CMD_CATEGORY_PUBSUB;        } else if (!strcasecmp(flag,"no-script")) {            c->flags |= CMD_NOSCRIPT;        } else if (!strcasecmp(flag,"random")) {            c->flags |= CMD_RANDOM;        } else if (!strcasecmp(flag,"to-sort")) {            c->flags |= CMD_SORT_FOR_SCRIPT;        } else if (!strcasecmp(flag,"ok-loading")) {            c->flags |= CMD_LOADING;        } else if (!strcasecmp(flag,"ok-stale")) {            c->flags |= CMD_STALE;        } else if (!strcasecmp(flag,"no-monitor")) {            c->flags |= CMD_SKIP_MONITOR;        } else if (!strcasecmp(flag,"no-slowlog")) {            c->flags |= CMD_SKIP_SLOWLOG;        } else if (!strcasecmp(flag,"cluster-asking")) {            c->flags |= CMD_ASKING;        } else if (!strcasecmp(flag,"fast")) {            c->flags |= CMD_FAST | CMD_CATEGORY_FAST;        } else if (!strcasecmp(flag,"no-auth")) {            c->flags |= CMD_NO_AUTH;        } else if (!strcasecmp(flag,"may-replicate")) {            c->flags |= CMD_MAY_REPLICATE;        } else {            /* Parse ACL categories here if the flag name starts with @. */            uint64_t catflag;            if (flag[0] == '@' &&                (catflag = ACLGetCommandCategoryFlagByName(flag+1)) != 0)            {                c->flags |= catflag;            } else {                sdsfreesplitres(argv,argc);                return C_ERR;            }        }    }

具體命令比較多

struct redisCommand redisCommandTable[] = {    {"module",moduleCommand,-2,     "admin no-script",     0,NULL,0,0,0,0,0,0},    {"get",getCommand,2,     "read-only fast @string",     0,NULL,1,1,1,0,0,0},}

以 set 為例子 {“set”,setCommand,-3, “write use-memory @string”, 0,NULL,1,1,1,0,0,0}

參考資料

• 《Redis 設(shè)計(jì)與實(shí)現(xiàn)》黃健宏


作者：老鄭 運(yùn)維開(kāi)發(fā)故事




歡迎關(guān)注：運(yùn)維開(kāi)發(fā)故事