<commit_before>---
layout: post
permalink: /blog/disque.html
title: "Disque实现分析"
category: articles
tags: [Jekyll]
---

    kiterunner_t
    TO THE HAPPY FEW


Disque是Redis作者正在开发中的一个分布式消息队列。

由于Disque基于Redis，复用了Redis的很多基础设施，如基本数据结构（容器）、网络服务、协议解析、命令和配置管理等。整个代码布局和风格也完全沿用了Redis的特征。因此，对于熟悉Redis代码的人来说，很容易上手；本文在了解Disque实现过程中，也就忽略那些来自Redis的部分，有兴趣的可以去参考关于Redis实现的分析。

本文主要内容包括以下几个功能点的实现：集群管理、客户端管理、job和queue管理、AOF。由于5.1假期时间所限，并没有使用Visio重新进行绘图，而是将阅读代码的时候手绘的图片直接传上来，后续等Disque正式版本发布后，再更新并补上。

使用3个节点进行代码分析，使用Java客户端Jedisque进行了一些客户端基本功能了解。

Disque设计为单个节点就是一个简单的事件循环处理服务器，尽管单个事件是串行执行的，但由于多个事件相互交织，各子模块的资源管理也存在一些复杂性，因此阅读代码的时候，需要时刻注意单个服务器中的事件流，如下：

* beforeSleep事件循环睡眠前的处理。
* serverCron，1ms的周期性定时器，该定时器包含了许多子模块需要进行定时处理的机制，如集群、客户端等等。
* processJobs，1ms的周期性定时器。
* acceptTcpHandler/aceptUnixHandler，接受来自客户端的连接处理。
* readQueryFromClient，响应客户端的命令。
* clusterAcceptHandler，集群握手中建立连接。
* clusterReadHandler，集群消息处理。
* bioProcessBackgroundJobs，后台线程处理。虽然这里起了一个多线程，但Disque只使用它来执行一些耗时的文件close、sync操作，并不会过多影响服务器状态，因此依然可以当做单进程单线程服务器。

## 1 集群管理

Disque的集群是全Mesh的，所有节点无主次之分，都全部直接相连，而且是双方都会进行握手建立两个独立的TCP连接。单节点的基本数据结构如下图所示：

![disque-cluster-data][1]


3个集群节点的互联关系如下图所示：

![disque-cluster-relation][2]

集群的主要功能分为两个部分：集群的管理，包括集群握手、节点失败监控；集群间job和queue的管理。本小节只描述集群自身的管理，而对于job和queue的分析放在相应的job和queue部分。

握手过程用于建立集群，可以有两种方式触发：cluster meet命令和gossip消息触发。下图所示为cluter meet命令触发的消息流，gossip也与此类似。

![disque-cluster-handshake][3]


* STEP1：a) 客户端发起cluster命令，接收到该命令的节点进行握手初始化，在clusterStartHandshake中，先创建一个集群节点数据结构（注意，此时节点ID是随机生成的，并不知道节点2的真实ID），并将其设置为DISQUE_NODE_HANDSHAKE和DISQUE_NODE_MEET标志。并响应客户端命令处理结果。换句话说，该命令只是触发了集群的握手过程，并不代表握手已经成功。b) 在cluster	Cron中，向目标节点2发起TCP连接。
* STEP2：节点2接收到来自节点1的连接请求，创建一个clusterLink，该link并未附着在任何地方，仅用于该socket事件处理的privdata中，如图3中A12连接中偏向于节点2的那个clusterLink所示。
* STEP3：节点1已经成功连接到节点2，创建一个clusterLink，并附着于clusterNode，如图3中A12所示。向节点2发起一个CLUSTERMSG_TYPE_P消息，去掉DISQUE_NODE_MEET标志。
* STEP4：节点2收到节点1的meet消息，创建对应的节点数据结构（节点ID也是随机生成的），同时也向节点1发起ping（此时并不会发meet消息，创建节点时没有设置DISQUE_NODE_MEET标志）。处理完成后，回给节点1一个CLUSTERMSG_TYPE_PONG消息。
* STEP5：节点1接收到节点2的pong消息，将节点2的ID修改成其真实ID，完成节点1自身的握手，并去掉DISQUE_NODE_HANDSHAKE标志。在随后的clusterCron中，向节点2发起一个ping消息。
* STEP6：接收到节点1的pong，将STEP4中随机生成的节点1的ID修改成真实的ID，去掉DISQUE_NODE_HANDSHAKE标志，节点2完成握手。

值得注意的是，在STEP4中，有2种case需要考虑，若节点2在随后的clusterCron向节点1发起ping，节点1的pong先到节点2，即这里正常握手的情况，完成握手。另一种情况是，节点1在STEP5中收到pong后，自身完成握手，随后在clusterCron中向节点2发起一个ping，此时节点2基本不干什么，直接回一个pong就好。

集群节点容错检测是在clusterCron中进行的，先是检测到某个节点的ping长时间没有响应，则转换为pfail状态，在消息处理过程中会进行检测是否已经达到fail计数了（超过集群节点数目加1），则转换为fail。

## 2 客户端管理

Disque客户端管理分为以下几个部分：

* 接受来自客户端的连接请求，服务器分配和初始化资源准备为客户端服务，加入到server.clients链表中，acceptTcpHandler完成这部分工作；当来自客户端的请求过多时，会直接拒绝请求。
* 客户端连接成功后，若有请求发过来，则会在readQueryFromClient中进行响应处理。这部分最终结果是处理各子模块的命令，但也有一些客户端控制的命令。
* 服务器在事件循环睡眠前的处理beforeSleep。
* 服务器周期性定时器处理。

为了实现上述目的，客户端管理的基本数据结构如下图所示。在图中需要注意以下几个数据结构：
全局容器记录了所有客户端信息server.clients
空闲连接管理和准备关闭的客户端放在server.clients_to_close链表
阻塞后需要防止为非阻塞的客户端放在server.unblocked_clients
server.monitors

![disque-client][4]


先看看与客户端有关的命令：monitor、client。

monitor：当某个客户端执行了monitor后，其他所有客户端的命令（非load以及带有M标记的命令）的相关信息都会在该客户端进行监视。在任何客户端执行大多数命令的时候，会调用replicationFeedMonitors将这些信息反馈给监视客户端。

client命令用于显示和修改客户端信息、控制客户端。各子命令的含义都一目了然，如下：
client list
client getname
client setname \<name>
client kill \<ip:port>
client kill \<option> [value] … \<option> [value]
client pause

这里关注一下kill和pause。对于kill而言，找到对应的客户端（可能有多个），直接释放所有与该客户端相关的资源，关闭客户端，并不是放在server.clients_to_close链表中。

pause命令执行后，所有客户端都会挂住，这是通过一个全局变量server.clients_paused标志来实现。在接收到客户端请求时以及serverCron中会通过函数clientsArePaused检查该标志，若已经超过了pause时间，则清除该标志，并将所有客户端server.clients都添加到server.unblocked_clients中，

阻塞客户端的几种情况：

* 某个客户端执行了client pause命令后，当pause时间到后，那么所有客户端都会添加到server.unblocked_clients链表中；
* 某个客户端执行了命令，等待复制到其他节点，此时client->btype为DISQUE_BLOCKED_JOB_REPL；
* 客户端在queue上执行了命令，但该queue没有对应的job或queue不存在，此时client->btype为DISQUE_BLOCKED_QUEUES。

对于处于阻塞的客户端，Disque会在以下几个时间点处理这些从阻塞变为非阻塞的客户端：

* 事件睡眠前的beforeSleep->processUnblockedClients
* 服务器轮询的clientsCron->clientsCronHandleTimeout
* 命令处理完成后执行proccessCommand->handleC
* 集群中复制完成后jobReplicationAchieved

## 3 job和queue管理


## 4 AOF



[1]: /images/disque/disque-cluster-data.png "disque-cluster-data"
[2]: /images/disque/disque-cluster-relation.png "disque-cluster-relation"
[3]: /images/disque/disque-cluster-handshake.png "disque-cluster-handshake"
[4]: /images/disque/disque-client.png "disque-client"<commit_msg>纠正图片格式<commit_after>---
layout: post
permalink: /blog/disque.html
title: "Disque实现分析"
category: articles
tags: [Jekyll]
---

    kiterunner_t
    TO THE HAPPY FEW


Disque是Redis作者正在开发中的一个分布式消息队列。

由于Disque基于Redis，复用了Redis的很多基础设施，如基本数据结构（容器）、网络服务、协议解析、命令和配置管理等。整个代码布局和风格也完全沿用了Redis的特征。因此，对于熟悉Redis代码的人来说，很容易上手；本文在了解Disque实现过程中，也就忽略那些来自Redis的部分，有兴趣的可以去参考关于Redis实现的分析。

本文主要内容包括以下几个功能点的实现：集群管理、客户端管理、job和queue管理、AOF。由于5.1假期时间所限，并没有使用Visio重新进行绘图，而是将阅读代码的时候手绘的图片直接传上来，后续等Disque正式版本发布后，再更新并补上。

使用3个节点进行代码分析，使用Java客户端Jedisque进行了一些客户端基本功能了解。

Disque设计为单个节点就是一个简单的事件循环处理服务器，尽管单个事件是串行执行的，但由于多个事件相互交织，各子模块的资源管理也存在一些复杂性，因此阅读代码的时候，需要时刻注意单个服务器中的事件流，如下：

* beforeSleep事件循环睡眠前的处理。
* serverCron，1ms的周期性定时器，该定时器包含了许多子模块需要进行定时处理的机制，如集群、客户端等等。
* processJobs，1ms的周期性定时器。
* acceptTcpHandler/aceptUnixHandler，接受来自客户端的连接处理。
* readQueryFromClient，响应客户端的命令。
* clusterAcceptHandler，集群握手中建立连接。
* clusterReadHandler，集群消息处理。
* bioProcessBackgroundJobs，后台线程处理。虽然这里起了一个多线程，但Disque只使用它来执行一些耗时的文件close、sync操作，并不会过多影响服务器状态，因此依然可以当做单进程单线程服务器。

## 1 集群管理

Disque的集群是全Mesh的，所有节点无主次之分，都全部直接相连，而且是双方都会进行握手建立两个独立的TCP连接。单节点的基本数据结构如下图所示：

![disque-cluster-data][1]


3个集群节点的互联关系如下图所示：

![disque-cluster-relation][2]

集群的主要功能分为两个部分：集群的管理，包括集群握手、节点失败监控；集群间job和queue的管理。本小节只描述集群自身的管理，而对于job和queue的分析放在相应的job和queue部分。

握手过程用于建立集群，可以有两种方式触发：cluster meet命令和gossip消息触发。下图所示为cluter meet命令触发的消息流，gossip也与此类似。

![disque-cluster-handshake][3]


* STEP1：a) 客户端发起cluster命令，接收到该命令的节点进行握手初始化，在clusterStartHandshake中，先创建一个集群节点数据结构（注意，此时节点ID是随机生成的，并不知道节点2的真实ID），并将其设置为DISQUE_NODE_HANDSHAKE和DISQUE_NODE_MEET标志。并响应客户端命令处理结果。换句话说，该命令只是触发了集群的握手过程，并不代表握手已经成功。b) 在cluster	Cron中，向目标节点2发起TCP连接。
* STEP2：节点2接收到来自节点1的连接请求，创建一个clusterLink，该link并未附着在任何地方，仅用于该socket事件处理的privdata中，如图3中A12连接中偏向于节点2的那个clusterLink所示。
* STEP3：节点1已经成功连接到节点2，创建一个clusterLink，并附着于clusterNode，如图3中A12所示。向节点2发起一个CLUSTERMSG_TYPE_P消息，去掉DISQUE_NODE_MEET标志。
* STEP4：节点2收到节点1的meet消息，创建对应的节点数据结构（节点ID也是随机生成的），同时也向节点1发起ping（此时并不会发meet消息，创建节点时没有设置DISQUE_NODE_MEET标志）。处理完成后，回给节点1一个CLUSTERMSG_TYPE_PONG消息。
* STEP5：节点1接收到节点2的pong消息，将节点2的ID修改成其真实ID，完成节点1自身的握手，并去掉DISQUE_NODE_HANDSHAKE标志。在随后的clusterCron中，向节点2发起一个ping消息。
* STEP6：接收到节点1的pong，将STEP4中随机生成的节点1的ID修改成真实的ID，去掉DISQUE_NODE_HANDSHAKE标志，节点2完成握手。

值得注意的是，在STEP4中，有2种case需要考虑，若节点2在随后的clusterCron向节点1发起ping，节点1的pong先到节点2，即这里正常握手的情况，完成握手。另一种情况是，节点1在STEP5中收到pong后，自身完成握手，随后在clusterCron中向节点2发起一个ping，此时节点2基本不干什么，直接回一个pong就好。

集群节点容错检测是在clusterCron中进行的，先是检测到某个节点的ping长时间没有响应，则转换为pfail状态，在消息处理过程中会进行检测是否已经达到fail计数了（超过集群节点数目加1），则转换为fail。

## 2 客户端管理

Disque客户端管理分为以下几个部分：

* 接受来自客户端的连接请求，服务器分配和初始化资源准备为客户端服务，加入到server.clients链表中，acceptTcpHandler完成这部分工作；当来自客户端的请求过多时，会直接拒绝请求。
* 客户端连接成功后，若有请求发过来，则会在readQueryFromClient中进行响应处理。这部分最终结果是处理各子模块的命令，但也有一些客户端控制的命令。
* 服务器在事件循环睡眠前的处理beforeSleep。
* 服务器周期性定时器处理。

为了实现上述目的，客户端管理的基本数据结构如下图所示。在图中需要注意以下几个数据结构：
全局容器记录了所有客户端信息server.clients
空闲连接管理和准备关闭的客户端放在server.clients_to_close链表
阻塞后需要防止为非阻塞的客户端放在server.unblocked_clients
server.monitors

![disque-client][4]


先看看与客户端有关的命令：monitor、client。

monitor：当某个客户端执行了monitor后，其他所有客户端的命令（非load以及带有M标记的命令）的相关信息都会在该客户端进行监视。在任何客户端执行大多数命令的时候，会调用replicationFeedMonitors将这些信息反馈给监视客户端。

client命令用于显示和修改客户端信息、控制客户端。各子命令的含义都一目了然，如下：
client list
client getname
client setname \<name>
client kill \<ip:port>
client kill \<option> [value] … \<option> [value]
client pause

这里关注一下kill和pause。对于kill而言，找到对应的客户端（可能有多个），直接释放所有与该客户端相关的资源，关闭客户端，并不是放在server.clients_to_close链表中。

pause命令执行后，所有客户端都会挂住，这是通过一个全局变量server.clients_paused标志来实现。在接收到客户端请求时以及serverCron中会通过函数clientsArePaused检查该标志，若已经超过了pause时间，则清除该标志，并将所有客户端server.clients都添加到server.unblocked_clients中，

阻塞客户端的几种情况：

* 某个客户端执行了client pause命令后，当pause时间到后，那么所有客户端都会添加到server.unblocked_clients链表中；
* 某个客户端执行了命令，等待复制到其他节点，此时client->btype为DISQUE_BLOCKED_JOB_REPL；
* 客户端在queue上执行了命令，但该queue没有对应的job或queue不存在，此时client->btype为DISQUE_BLOCKED_QUEUES。

对于处于阻塞的客户端，Disque会在以下几个时间点处理这些从阻塞变为非阻塞的客户端：

* 事件睡眠前的beforeSleep->processUnblockedClients
* 服务器轮询的clientsCron->clientsCronHandleTimeout
* 命令处理完成后执行proccessCommand->handleC
* 集群中复制完成后jobReplicationAchieved

## 3 job和queue管理


## 4 AOF



[1]: /images/disque/disque-cluster-data.jpg "disque-cluster-data"
[2]: /images/disque/disque-cluster-relation.jpg "disque-cluster-relation"
[3]: /images/disque/disque-cluster-handshake.jpg "disque-cluster-handshake"
[4]: /images/disque/disque-client.jpg "disque-client"
{'fraction_non_alphanumeric': 0.08802694228312623, 'fraction_numerical': 0.014400185808849147, 'mean_word_length': 15.529750479846449, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4611887', 'n_tokens_mistral': 5946, 'n_tokens_neox': 6879, 'n_words': 227}