前言

前置知识

  • Redis 虽然是单线程的，但是它利用了内核的 IO 多路复用，从而能同时监听多个连接
  • Redis6 出现了可以利用多个 IO 线程并发进行的操作

那么问题来了，这两者会导致我们的分布式锁的原子性有影响吗？

我们知道当我们使用 redis 作为分布式锁的时候，通常会使用 SET key value EX 10 NX 命令来加锁，获得锁的客户端才能成功 SET 这个 key，那么问题来了，这条命令在多线程的情况下是一个原子操作吗？

其实答案是显而易见的，因为 redis 的设计者肯定考虑到了向前兼容的问题，并且也不会让这样的特性消失，所以在问这个问题以前，我虽然不能肯定，但是还是能自信的回答，但没有足够的底气。 今天的目标就是找到真正的原因。

问题的两个方面

上锁

上锁，没啥多说的直接 SET key value EX 10 NX 就可以了

解锁

解锁，有两种：

  • 一种是客户端自行保证锁只有自己拿自己解，那么直接让自己去 DEL 就可以了
  • 另一种是不信任客户端，那么可以使用 lua 脚本，先通过 get 确定对应 key 的值是否正确，如果正确再 del，整个 lua 脚本通过 EVAL 执行

只要上锁和解锁操作都能保证，就能解决问题。

执行命令的过程

那么问题的关键就是命令的执行过程，Redis 执行命令也是需要有过程的，客户端一个命令过来，不会直接就啪的执行了，而是有很多前置条件和步骤。

大致可分为：

  1. 读取
  2. 解析
  3. 执行
  4. 返回

其中，命令读取和解析显然是不会影响数据的，所以当然多线程执行也没有问题。最关键的步骤也就是执行了。

IO 多路复用

先来看看 IO 多路复用会有影响吗？

代码来自： https://github.com/redis/redis/blob/074e28a46eb2646ab33002731fac6b4fc223b0bb/src/ae_epoll.c#L109

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval, numevents = 0;

    retval = epoll_wait(state->epfd,state->events,eventLoop->setsize,
            tvp ? (tvp->tv_sec*1000 + (tvp->tv_usec + 999)/1000) : -1);
    if (retval > 0) {
        int j;

        numevents = retval;
        for (j = 0; j < numevents; j++) {
            int mask = 0;
            struct epoll_event *e = state->events+j;

            if (e->events & EPOLLIN) mask |= AE_READABLE;
            if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
            if (e->events & EPOLLERR) mask |= AE_WRITABLE|AE_READABLE;
            if (e->events & EPOLLHUP) mask |= AE_WRITABLE|AE_READABLE;
            eventLoop->fired[j].fd = e->data.fd;
            eventLoop->fired[j].mask = mask;
        }
    } else if (retval == -1 && errno != EINTR) {
        panic("aeApiPoll: epoll_wait, %s", strerror(errno));
    }

    return numevents;
}

没事，不要担心看不懂，只要抓住最关键的地方 epoll_wait 这个我们很熟悉对吧，我们就可以看到这里一次循环拿出了一组 events，这些事件都是一股脑儿过来的。

其实 IO 多路复用本身没有问题，无论是 select 还是 epoll 只是将所有的 socket 的 fd 做了一个集合而已，而告诉你那些 fd 出现了事件，让你具体去处理。如果你不愿意多线程处理这些读写事件，那么 IO 多路复用是不会逼你的。

多线程

多线程倒是真的有可能会出问题。那如果我们自己去考虑实现的话，当一个命令被多线程去同时执行，那势必会有竞争，所以我们为了尽可能利用多线程去加速，也只能加速，命令接收/解析/返回执行结果的部分。故，其实 Redis 的设计者也只是将多线程运用到了执行命令的前后。

代码在： https://github.com/redis/redis/blob/4ba47d2d2163ea77aacc9f719db91af2d7298905/src/networking.c#L2465

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
int processInputBuffer(client *c) {
    /* Keep processing while there is something in the input buffer */
    while(c->qb_pos < sdslen(c->querybuf)) {
        /* Immediately abort if the client is in the middle of something. */
        if (c->flags & CLIENT_BLOCKED) break;

        /* Don't process more buffers from clients that have already pending
         * commands to execute in c->argv. */
        if (c->flags & CLIENT_PENDING_COMMAND) break;

        /* Don't process input from the master while there is a busy script
         * condition on the slave. We want just to accumulate the replication
         * stream (instead of replying -BUSY like we do with other clients) and
         * later resume the processing. */
        if (isInsideYieldingLongCommand() && c->flags & CLIENT_MASTER) break;

        /* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is
         * written to the client. Make sure to not let the reply grow after
         * this flag has been set (i.e. don't process more commands).
         *
         * The same applies for clients we want to terminate ASAP. */
        if (c->flags & (CLIENT_CLOSE_AFTER_REPLY|CLIENT_CLOSE_ASAP)) break;

        /* Determine request type when unknown. */
        if (!c->reqtype) {
            if (c->querybuf[c->qb_pos] == '*') {
                c->reqtype = PROTO_REQ_MULTIBULK;
            } else {
                c->reqtype = PROTO_REQ_INLINE;
            }
        }

        if (c->reqtype == PROTO_REQ_INLINE) {
            if (processInlineBuffer(c) != C_OK) break;
        } else if (c->reqtype == PROTO_REQ_MULTIBULK) {
            if (processMultibulkBuffer(c) != C_OK) break;
        } else {
            serverPanic("Unknown request type");
        }

        /* Multibulk processing could see a <= 0 length. */
        if (c->argc == 0) {
            resetClient(c);
        } else {
            /* If we are in the context of an I/O thread, we can't really
             * execute the command here. All we can do is to flag the client
             * as one that needs to process the command. */
            if (io_threads_op != IO_THREADS_OP_IDLE) {
                serverAssert(io_threads_op == IO_THREADS_OP_READ);
                c->flags |= CLIENT_PENDING_COMMAND;
                break;
            }

            /* We are finally ready to execute the command. */
            if (processCommandAndResetClient(c) == C_ERR) {
                /* If the client is no longer valid, we avoid exiting this
                 * loop and trimming the client buffer later. So we return
                 * ASAP in that case. */
                return C_ERR;
            }
        }
    }

同样的，也不用慌，抓住重点的部分

  • 当出现 CLIENT_PENDING_COMMAND 状态的时候是直接 break 的，后面就根本不处理，而这个状态就是表示客户端当前正在等待执行的命令。在这个状态下，客户端不能发送其他命令，直到当前命令的执行结果返回。
  • 最终执行命令是在 processCommandAndResetClient 方法

总结

总结一下，IO 多路复用本身其实没有影响，而 Redis 真正执行命令的前后利用多线程来加速，加速命令的读取和解析，加速将执行结果返回客户端。所以，本质上 “IO多路复用和多线程会影响Redis分布式锁吗？” 而这个问题与分布式锁其实没有必然联系，分布式锁本质其实也是执行一条命令。故，其实面试官问这个问题的原因更多的是关心你对 IO 多路复用和多线程在 Redis 实践的理解。