嘿呀~

是的，在经过一堆烦人考试的折磨后，咱又回来啦！虽然还有一科遗传学，不过先不管啦！

太臃肿啦！

是的，cuscuta的工作队列设计模式，我说的是结果队列的设计模式——太过于臃肿啦！所以我打算要稍微重构一下这玩意……

重构

旧的redis模式为：

1
2
3
4
5
6
7
8
9
10
11
cuscuta:
	pending:
		index:
			[JobUid [JobTag]]
	results:
		index:
			[JobUid [JobTag]]
		value:
			[JobUid [Result]]
	jobs:
		[Segment descriptor [Jobs]]

在思考后，我发选这个模式存在以下缺陷：

  • 维护了pending和results两个JobTag列表
  • 这浪费了两倍的空间
  • 如果需要统计任务的精确完成情况，就需要查询两次JobTag列表，不易统计任务的完成情况
  • 最重要的是，目前由于只有两个index和value，共计三个需要维护的列表，如果需要给任务添加错误回传，就需要添加第四个维护的列表，这在技术上可行，但是复杂的redis结构逻辑会加剧我的心智负担，并且也不利于后续的扩展（如果我要加第五个字段，阁下又该如何应对）

所以，我打算将redis模式改成下面这样的：

1
2
3
4
5
6
7
cuscuta:
	status:
		Hash<JobUid, [ImprovedJobTag]>
	results:
		Hash<JobUid, [Result]>
	jobs:
		[Segment descriptor[Jobs]]

别急，我当然不会用 ImprovedJobTag 这种名字……

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
/// 更改后的 JobTag
#[derive(Debug, Serialize, Deserialize)]
pub struct JobTag {
    /// 添加这个，用于记录 Job 当前状态，同样要更改 ETA 计算方式
    pub status: JobStatus,
    
    /// 将 `last_job_id` 改为 `job_ids` ，统计使用过的 `job_id` 数目，就可统计出重试次数
    pub job_ids: Vec<String>,
    
    /// 同样，雷打不动的分片队列信息
    pub queue: SubQueue,
    
    /// Job 的关键信息，同样必需
    pub job_essential: JobEssential,
    
    /// 添加这个，用于记录 Job 的失败信息
    pub failures: Vec<JobFailure>,
}

/// Job 的工作状态
#[derive(Debug, Serialize, Deserialize)]
pub enum JobStatus {
    Queueing,
    Pending,
    Failed,
}

/// Job 的错误信息
#[derive(Debug, Serialize, Deserialize)]
pub struct JobFailure {
    /// Job 失败的原因
    pub fail_type: JobFailureType,
    
    /// Job 失败的时间
    pub timestamp: u64
}

/// Job 失败的类型
#[derive(Debug, Serialize, Deserialize, thiserror::Error)]
pub enum JobFailureType {
    /// 例如，好友找不到
    #[error("friend not found")]
    FriendNotFound,
    
    /// ... Add more
}

同时，多List简单追加信息改为使用JobUid搜索后更改，我承认这可能会降低性能，但是我认为这点性能损失不在话下.unwrap()

显然，我需要更改worker的逻辑，使其在更新本地缓存任务状态时也更新redis数据库里的

“You can’t be more careful about error handing”

我承认cuscuta现在的错误处理还有待提升，尤其是核心业务逻辑中极有可能出现的错误，我还放到现在才处理

当然，我也不敢保证除了这个以外我没有别的错误还没考虑到

如果上述更改凑效，我就可以对错误进行分级：一些错误是不可容忍的，例如“好友找不到”，遇到了这种错误，worker在清理任务时，就需要将其任务状态直接置为Failed，而不是置为Queueing然后重入队

说到这里，我觉得我也有必要修改一下目前的worker循环——目前worker循环有滥用?表达式的倾向，而实际情况是，并不是所有 Result::Err 都是需要直接中断掉工作循环的：

  • 一些错误只影响任务本身（例如无法添加好友），这时候就需要记录这个错误，并且按照严重性在任务清理步骤中进行对应操作，例如写Failed，然后直接XACK，不重入队
  • 对于一些可以重试，但是这个worker本身已经无力回天的错误（例如读取单个好友信息失败——这个错误不常见，但是我不保证它不会发生），就需要写Queueing，然后重入队
  • 对于一些可以重试，只是因为网络问题的错误，就让worker一直重试——既然都有网络问题，这时候多半是redis数据库都连不上，要么是机房断网——在这种物理胁迫下，任何故障转移没有意义，不如不进行故障转移
  • 当然，如果worker遭遇不测，优雅停机了，worker就应该把自己未完成的任务全部重入队，并且失败信息加入一个”Worker halting”之类的；如果没有优雅停机，就等着别的worker xautoclaim吧，反正已经死透了，记得让通过xautoclaim领取的任务再加一个”Worker halting”之类的错误信息

以上对工作循环的重构提出了一些挑战，不过我想我可以解决

Dashboard……的前置

接下来让我们画一点大饼，如果我以后需要给cuscuta还是别的什么分布式集群添加一个可视化仪表盘，我需要暴露些什么东西呢？

一个简单的方法是：直接读取redis数据库的内容，然后将其解析成仪表盘上的数据

但是这显然不太优雅，尤其是如果我以后需要在一个仪表盘中查看多个集群（或者说，多个部署）的运行情况的话，我总不能让仪表盘做一堆适配吧？

所以，我觉得整一个统一的API规范是有必要的——每个部署内部维护一个简单的Deployment，这个Deployment只是读redis队列，或者是其它什么东西，然后将其映射为一个结构化的输出，前端只需要调用这些接口，解析这些统一结构的输出就可以了，简单来讲，把数据对接的操作丢给了部署本身，而不是Dashboard

// Dashboard API 正在设计中