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

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

阅读数: 次评论数: 0次 2023-06-21

× 文章目录

1. 代码全景
2. 一些技术实现
3. 一次请求的处理流程
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函数中初始化的
事件驱动消息流转：多个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缓冲池

一次请求的处理流程

三种连接

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

三个队列

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

用户通过 proxy connection 建立连接，产生一个 client connection
client connection 开始读取用户的请求数据，并将完整的请求根据 key 和设置的哈希规则选择 server, 然后将这个请求存放到 server 的 imsgq

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

// 把后端应答回来的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);
}

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

连接的回调函数

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发起连接请求

client发起命令请求

server连接可写事件触发

server返回回包触发可读事件

client连接可写事件触发

参考文章

文章地址	说明
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 国际许可协议进行许可。转载请注明出处！

twemproxy