celery 远程控制实现之 revoke

介绍

由于对 celery 在消息队列上实现远程控制的好奇，因而以 revoke 命令（取消任务)为例来研究其中的实现。

源码分析

1. worker 端对 revoke 的实现
   image.png

• 调用流: 任务的 revoke --> request 的 terminate --> 进程/线程 pool 的 terminate --> 最终本质还是调用 os.kill(pid, signal)

2. worker 端 revoke 在消息队列上的实现
   image.png

• 调用流：revoke 注册到 Panel 类的类变量 data --> Panel 的 data 赋值给 Mailbox.Node 的 handlers, worker 通过 on_message 来处理消息队列中的消息，实际上调用的 Mailbox.Node（TODO: 有关 kombu/pibox 的实现） 的 handle_message。
• 代码组织理解：注册函数的方式使得添加和减少控制器的函数不具有侵入性（不会改动源代码），并将控制器封装赋值给handlers。
• 注册函数的实现：函数装饰器结合返回类方法实现注册函数的方式值得参考，使用类变量和类方法来实现全局注册，在需要注册隔离时可以采用实例方法，这里通过 type 字段区分不同注册功能函数。

3. client 端 revoke 在消息队列上的实现
   image.png

• 调用流：client 调用 control 的 revoke 方法 --> 调用 control 的 broadcast 方法 --> 调用 Mailbox 的 _broadcast 方法 --> 调用 Mailbox 的 _publish 方法。