redis集群代理twemproxy源码阅读总结

字数统计: 2.7k字   |   阅读时长: 11分钟

阅读数: 15次    评论数: 0次
× 文章目录
  1. 1. 代码全景
  2. 2. 一些技术实现
  3. 3. 一次请求的处理流程
    1. 3.1. 三种连接
    2. 3.2. 三个队列
    3. 3.3. 连接的回调函数
    4. 3.4. 请求处理流转
      1. 3.4.1. client发起连接请求
      2. 3.4.2. client发起命令请求
      3. 3.4.3. server连接可写事件触发
      4. 3.4.4. server返回回包触发可读事件
      5. 3.4.5. client连接可写事件触发
  4. 4. 参考文章

阅读twemproxy的总结文章,方便后续复看和复用其中的技术

代码全景

源码笔记：https://github.com/damotiansheng/Reading-and-comprehense-twemproxy0.5.0.git

代码规模不大只有1w多行，而且功能划分的比较清晰，包括：

  • 事件处理： event/nc_epoll.c、event/nc_event.h、event/nc_evport.c、event/nc_kqueue.c
  • 各种Hash函数： hashkit/nc_crc16.c、hashkit/nc_crc32.c、hashkit/nc_fnv.c、hashkit /nc_hashkit.h、hashkit/nc_hsieh.c、hashkit/nc_jenkins.c、hashkit /nc_ketama.c、hashkit/nc_md5.c、hashkit/nc_modula.c、hashkit/nc_murmur.c、 -hashkit/nc_one_at_a_time.c、hashkit/nc_random.c
  • 协议： proto/nc_memcache.c、proto/nc_proto.h、proto/nc_redis.c
  • 自定义的数据类型： nc_array.c、nc_array.h、nc_string.c、nc_string.h
  • 网络通信相关： nc_connection.c、nc_connection.h、nc_client.c、nc_client.h、nc_proxy.c、nc_proxy.h
  • 信号处理： nc_signal.c、nc_signal.h
  • 关键数据结构和算法： nc_rbtree.h、nc_rbtree.c、nc_queue.h、nc_request.c、nc_response.c、nc_mbuf.c、 nc_mbuf.h、- nc_message.c、nc_message.h、nc_server.c、nc_server.h
  • 统计、日志和工具： nc_stats.c、nc_stats.h、nc_log.c、nc_log.h、nc_util.c、nc_util.h
  • 配置文件： nc_conf.c、nc_conf.h
  • 主程序： nc.c、nc_core.c、nc_core.h

一些技术实现

  • 采用epoll + 回调的单线程工作模式，单线程epoll监听，每个事件触发则调用相应的回调函数
  • 可监听多个地址，客户端向这些地址发起连接请求，每个地址将请求转发到后端的server池

实现上是通过创建多个socket，监听多个地址，然后加入epoll监听中，单线程epoll_wait监听请求

  • 三种连接：proxy：监听客户端连接、client：处理客户端数据连接、server：后端server的连接，每种连接都会保存起读写回调函数，每个连接会用一个conn数据结构表示，相关的读写回调函数会记录在此，每新建一个连接时会加入到时间管理器的监听事件中（每个连接有一个fd），并将该conn赋值给epoll_event.data.ptr，以便事件触发时得到该conn，从而调用到相应的回调函数，proxy连接的回调函数是在conn_get_proxy函数中初始化的，client和server连接的回调函数是在conn_get函数中初始化的

    image.png

  • 事件驱动消息流转：多个in/out队列，消息根据当前的处理流程会进入到某个队列中，从后端server得到回包后，会将请求的msg和回包的msg关联起来，然后client连接找到相应的回包并发送给客户端，整个流程是事件驱动的

  • 总共2个线程，主线程和stats线程

  • stats线程实现原理：一个简单的http服务，即请求一个http的url，返回stats数据，实现原理：创建一个新的stats线程，epoll_create创建新的事件管理器，创建监听fd，加入epoll的监听事件中并设置回调函数，有请求来了调用相应的回调函数进行处理，默认30秒触发一次

  • 超时处理：请求后端server如果配置了超时，则会进行超时处理，具体是发送给后端请求后，如果规定时间没返回，会超时关闭客户端连接，实现上是通过红黑树实现，key是超时的时间，value是连接，每次epoll从红黑树得到最小的时间（最左节点）赋值给ctx->timeout，然后传入epoll_wait的第四个参数，epoll_wait(ep, event, nevent, timeout)，到达规定时间后epoll_wait会返回，实现对超时事件的处理

在redis中，超时事件的实现是通过一个双向链表，每次遍历得到最早超时事件，然后传入epoll_wait的参数中的

  • 事件驱动实现：提供了统一的上层接口，底层封装了epoll、kqueue、evport实现，跟redis事件驱动实现类似
  • 实现了一些核心数据结构和算法：红黑树、一致性hash算法（用来选择后端server，将请求路由到后端某个server去）、变长字符串、变长数组、队列（实现来源于https://github.com/freebsd/freebsd-src/blob/master/sys/sys/queue.h）、使用libyaml库解析这个yaml格式的配置文件
  • 守护进程实现：nc_daemonize函数
  • 连接缓冲池技术&连接的LRU管理：每个后端server维护一个s_conn_q连接队列，建立的连接放到该队列中，每次需要建立该server的连接就判断是否超过最大连接数（pool->server_connections，默认为1），没有则建立新的连接插入队列，超过则从队头取连接conn并插入到队尾（即LRU算法），然后返回这个连接，完成对连接的复用，代码见conn_get函数
  • 空闲连接队列free_connq，三种连接：proxy，client，server在释放连接时会将连接放到空闲连接队列中，需要建立连接时从该空闲连接队列取连接，而不是重新分配内存，相当于一个空闲连接池，三种连接复用该空闲连接池，简单理解就是一个连接数据结构的缓存池，不用每次重新分配
  • 消息结构的缓存池技术：读取消息需要先分配msg结构，优先从空闲msg队列free_msgq中获取，相当于一个msg缓冲池

一次请求的处理流程

三种连接

img

  • proxy connection， 用来监听用户建立连接的请求，建立连接成功后会对应产生一个客户端连接
  • client connection，由建连成功后产生，用户读写数据都是通过 client connection 解析请求后，根据 key 和哈希规则选择一个 server 进行转发
  • server connection，转发用户请求到缓存资源并接收和解析响应数据转回 client connection，client connection 将响应返回到用户

三个队列

img
上图的 client connection 之所以没有 imsgq 是因为请求解析完可以直接进入 server 的 imsgq

  1. 用户通过 proxy connection 建立连接，产生一个 client connection

  2. client connection 开始读取用户的请求数据，并将完整的请求根据 key 和设置的哈希规则选择 server, 然后将这个请求存放到 server 的 imsgq

  3. 接着 server connection 发送 imsgq 请求到远程资源，发送完成之后(写 tcp buffer) 就会将 msg 从 imsgq 迁移到 omsgq，响应回来之后从 omsgq 队列里面找到这个对应的 msg 以及 client connection

    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
    // 把后端应答回来的msg和客户端msg关联在一起,然后通过req_done函数里面的event_add_out，来触发客户端conn把这些后端回包msg发送出去
    // 参数msg是server的回包msg
    static void
    rsp_forward(struct context *ctx, struct conn *s_conn, struct msg *msg) {
        rstatus_t status;
        struct msg *pmsg;
        struct conn *c_conn;
        uint32_t msgsize;

        ASSERT(!s_conn->client && !s_conn->proxy);
        msgsize = msg->mlen;

        /* response from server implies that server is ok and heartbeating */
        server_ok(ctx, s_conn);

        /* dequeue peer message (request) from server */
        pmsg = TAILQ_FIRST(&s_conn->omsg_q);   // 取第一个队列元素，先前入队是用TAILQ_INSERT_TAIL(&conn->omsg_q, msg, s_tqe);
        ASSERT(pmsg != NULL && pmsg->peer == NULL);
        ASSERT(pmsg->request && !pmsg->done);

    	// 请求msg从server的输出队列中出队
        s_conn->dequeue_outq(ctx, s_conn, pmsg);  // 调用 req_server_dequeue_omsgq 函数，会调用TAILQ_REMOVE(&conn->omsg_q, msg, s_tqe);
        pmsg->done = 1;

        /* establish msg <-> pmsg (response <-> request) link */
    	// 核心逻辑：建立请求msg和回包msg的连接
    	/* pmsg为接收客户端报文的msg信息，参数msg为后端应答回来的msg信息 */
    	/* 把接收的客户端msg信息和后端应答回来的msg信息进行关联 */
    	// 如果2个client复用了同一个server连接，client1先发送了get a, client2接着发送了get b, 在server连接的omsg_q就是(client1_req, client2_req)，但是
    	// client2的响应先回来，即这里的msg是get b的结果，但是pmsg是取了第一个，即client1，岂不是有问题？
    	// 没有问题，因为复用同一个server连接，get a和get b是先后发送到同一个server，由于后端是redis，单线程处理，
    	// 所以get a(client1请求)回包也是先回来，get b（client2请求）的回包后回来，不会出现client2请求先回来的情况
        pmsg->peer = msg;
        msg->peer = pmsg;

        msg->pre_coalesce(msg);

        c_conn = pmsg->owner;  // 找到client的conn
        ASSERT(c_conn->client && !c_conn->proxy);

        if (req_done(c_conn, TAILQ_FIRST(&c_conn->omsg_q))) {
            status = event_add_out(ctx->evb, c_conn);  // 注册client连接c_conn的读写事件到事件管理器中进行监听
            if (status != NC_OK) {
                c_conn->err = errno;
            }
        }

        rsp_forward_stats(ctx, s_conn->owner, msg, msgsize);
    }

  4. 最后将响应内容放到 client connection 的 omsgq，由 client connection 将数据发送回客户端。

连接的回调函数

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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
proxy连接：
conn->recv = proxy_recv;
conn->recv_next = NULL;
conn->recv_done = NULL;

conn->send = NULL;
conn->send_next = NULL;
conn->send_done = NULL;

conn->close = proxy_close;
conn->active = NULL;

conn->ref = proxy_ref;
conn->unref = proxy_unref;


client连接：
conn->recv = msg_recv;
conn->recv_next = req_recv_next;
conn->recv_done = req_recv_done;

conn->send = msg_send;
conn->send_next = rsp_send_next;
conn->send_done = rsp_send_done;

conn->close = client_close;
conn->active = client_active;

conn->ref = client_ref;
conn->unref = client_unref;

conn->enqueue_inq = NULL;
conn->dequeue_inq = NULL;
conn->enqueue_outq = req_client_enqueue_omsgq;
conn->dequeue_outq = req_client_dequeue_omsgq;


server连接：
conn->recv = msg_recv;
conn->recv_next = rsp_recv_next;
conn->recv_done = rsp_recv_done;

conn->send = msg_send;
conn->send_next = req_send_next;
conn->send_done = req_send_done;

conn->close = server_close;
conn->active = server_active;

conn->ref = server_ref;
conn->unref = server_unref;

conn->enqueue_inq = req_server_enqueue_imsgq;
conn->dequeue_inq = req_server_dequeue_imsgq;
conn->enqueue_outq = req_server_enqueue_omsgq;
conn->dequeue_outq = req_server_dequeue_omsgq;


*             Client+             Proxy           Server+
 *                              (nutcracker)
 *                                   .
 *       msg_recv {read event}       .       msg_recv {read event}
 *         +                         .                         +
 *         |                         .                         |
 *         \                         .                         /
 *         req_recv_next             .             rsp_recv_next
 *           +                       .                       +
 *           |                       .                       |       Rsp
 *           req_recv_done           .           rsp_recv_done      <===
 *             +                     .                     +
 *             |                     .                     |
 *    Req      \                     .                     /
 *    ===>     req_filter*           .           *rsp_filter
 *               +                   .                   +
 *               |                   .                   |
 *               \                   .                   /
 *               req_forward-//  (a) . (c)  \\-rsp_forward
 *                                   .
 *                                   .
 *       msg_send {write event}      .      msg_send {write event}
 *         +                         .                         +
 *         |                         .                         |
 *    Rsp' \                         .                         /     Req'
 *   <===  rsp_send_next             .             req_send_next     ===>
 *           +                       .                       +
 *           |                       .                       |
 *           \                       .                       /
 *           rsp_send_done-//    (d) . (b)    //-req_send_done
 *
 *
 * (a) -> (b) -> (c) -> (d) is the normal flow of transaction consisting
 * of a single request response, where (a) and (b) handle request from
 * client, while (c) and (d) handle the corresponding response from the
 * server.

请求处理流转

前文说过三种连接，都注册到了事件管理器中，并保存了读写回调函数（proxy连接只有读函数，因为只需要accept接收client连接请求），当有相应的事件触发，就会调用相应的回调函数进行处理，一个请求的处理流程为：client发起连接请求 -> client发起命令请求 -> server连接可写 -> server连接可读 -> client连接可写，下面按这些事件依次介绍，可以看到msg使用到了三个队列（颜色标识）：

client发起连接请求

img

client发起命令请求

img

server连接可写事件触发

img

server返回回包触发可读事件

img

client连接可写事件触发

img

参考文章

文章地址 说明
https://www.cnblogs.com/foxmailed/p/3623817.html 源码解析，请求消息的流转过程和处理过程
https://idning.github.io/twemproxy-src.html 内容很全，各个介绍，源码分析
https://hulkdev.com/posts-meitu-opensource-twemproxy/ 美图多进程 twemproxy 开源实现：https://github.com/bitleak/twemproxy
https://juejin.cn/post/6844903981429293063 twemproxy源码分析，核心流程介绍
https://github.com/joeylichang/joeylichang.github.io/blob/master/src/redis/twemproxy.md Twemproxy介绍
https://xie.infoq.cn/article/2ee961483c66a146709e7e861 结合nginx架构优化twemproxy
https://xie.infoq.cn/article/2ee961483c66a146709e7e861 二次开发，Nginx master-worker机制引入到twemproxy
https://github.com/twitter/twemproxy/commit/0060f4a6467903ad66d7fe5721653c327ef9f675#diff-1345cb5f953e43b06356881448bf3cecd997604cfd2240f10e994113892ea88a twemproxy添加ping命令修改
本文作者： L
本文链接： http://damotiansheng.github.io/twemproxy/2023-06-21/redis集群代理twemproxy源码阅读总结.html
版权声明： 本作品采用 知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议 进行许可。转载请注明出处！
知识共享许可协议
赏

谢谢你请我吃糖果

支付宝
微信
本站总访问量4193次 | 本站访客数2351人 | 本站已安全运行 3190 天 00 小时 57 分 41 秒