Celery 原理

[pengwk]

问题：

• celery 入队一个任务是怎样的？
• celery 是怎样从队列里拿出一个任务并执行的？

https://github.com/celery/celery https://www.fullstackpython.com/celery.html

[pengwk]

入队的数据结构

http://docs.celeryproject.org/en/latest/internals/protocol.html#version-2

[pengwk]

https://liuliqiang.info/post/celery-source-analysis-worker-execute-engine

[pengwk]

https://gtcsq.readthedocs.io/en/latest/openstack/kombu.html

[pengwk]

运行的地方：https://sourcegraph.com/github.com/celery/celery@76d10453ab9267c45b12d7c60b5ee0e4113b4369/-/blob/celery/app/trace.py?utm_source=share#L383

[pengwk]

https://www.youtube.com/watch?v=ceJ-vy7fvus https://www.youtube.com/watch?v=kDoHrFLkahA https://www.youtube.com/watch?v=XjzyOyLbvN8 https://www.youtube.com/watch?v=Bo6UtRhedjE https://www.youtube.com/watch?v=E708csv4XgY

[pengwk]

用例及其对应的数据结构和算法

• revoke https://github.com/celery/celery/wiki/Revoking-tasks
• call
• terminate
• beat
• debug

[pengwk]

相关概念

• worker
• prefork
• eventlet
• greenlet
• pool
• remote-control
• auto-scaling
• ack_late 等任务完成后再 ack。 ack 的定义是什么？
• prefetch
• routing task
• work-flow
• heartbeat
• revoke
• restart worker
• monitor
• application a.k.a app
• task registry
• channel
• consumer
• backend
• task
• broadcast
• task timeout
• mailbox
• connection
• Request: that specifies how tasks are executed.
• MainProcess
• eventloop
• pool
• weakref
• drain_events

相互关联的概念

• task application

关联的组件开源项目

• https://github.com/Bahus/celery-prometheus-exporter
• https://github.com/zerok/celery-prometheus-exporter
• https://zhuanlan.zhihu.com/p/51671229
• flower
• red-beat https://github.com/sibson/redbeat
• 监控 部署 运维

[pengwk]

主要功能

[pengwk]

文章

• 使用 Celery

[pengwk]

怎么启动，怎么关闭，怎么监控？

[pengwk]

用到了哪些设计模式？

[pengwk]

heartbeat

• node
• gossip Worker <-> Worker communication Bootstep

[pengwk]

没有中心化的东西怎么相互确保还活着？ worker 具体指什么？一个进程？如果是，究竟是一个怎样的进程？

[pengwk]

worker

https://docs.celeryproject.org/en/latest/userguide/workers.html

https://linux.die.net/man/2/getrusage

System usage statistics. The fields available may be different on your platform.

From getrusage(2):

stime 系统调用时间

Time spent in operating system code on behalf of this process.

utime 花在用户指令上的时间

Time spent executing user instructions.

maxrss 这个进程最大的驻留内存

The maximum resident size used by this process (in kilobytes).

idrss 用于保存数据的不共享内存大小

Amount of non-shared memory used for data (in kilobytes times ticks of execution)

isrss 不共享用于堆栈的内存大小

Amount of non-shared memory used for stack space (in kilobytes times ticks of execution)

ixrss 与其他内存共享的内存数量。

Amount of memory shared with other processes (in kilobytes times ticks of execution).

inblock 这个进程读磁盘文件的次数

Number of times the file system had to read from the disk on behalf of this process.

oublock 这个进程写磁盘的次数

Number of times the file system has to write to disk on behalf of this process.

majflt I/O 时出现的页错误次数

Number of page faults that were serviced by doing I/O.

minflt 没有 I/O 时的页错误数

Number of page faults that were serviced without doing I/O.

msgrcv 收到的 IPC 消息数

Number of IPC messages received.

msgsnd 发送的 IPC 消息数

Number of IPC messages sent.

nvcsw 自愿发起 context switch 次数

Number of times this process voluntarily invoked a context switch.

nivcsw 非自愿发起 context switch 次数

Number of times an involuntary context switch took place.

nsignals 收到的信号数

Number of signals received.

nswap 进程完全交换到磁盘的次数

The number of times this process was swapped entirely out of memory.

[pengwk]

broadcast

https://github.com/celery/celery/blob/8e016e667ae16958043b096e5cb89ac0f7dd7989/celery/app/control.py#L443

[pengwk]

一个远程控制指令是怎么被执行的？

[pengwk]

为什么类变量和实例变量这样用？

[pengwk]

drain 排空耗尽

[pengwk]

Redis

command

• pmessage
• PUBLISH
• MULTI
• BRPOP
• ZREM
• SETEX
• ZADD https://redis.io/commands/zadd
• MULTI
• EXEC
• HDEL
• HSET
• WATCH
• UNWATCH
• TTL
• EXPIRES
• SETNX
• ZREVRANGEBYSCORE

key

• unacked_index
• unacked_mutex
• default
• unacked

http://xargin.com/redis-protocal/ https://redis.io/topics/protocol

[pengwk]

ack

[pid 17740] sendto(29, "*1\r\n$5\r\nMULTI\r\n*3\r\n$4\r\nZREM\r\n$13\r\nunacked_index\r\n$36\r\ne6bad363-8d7c-4a02-8a8e-2a4cc58dfdf7\r\n*3\r\n$4\r\nHDEL\r\n$7\r\nunacked\r\n$36\r\ne6bad363-8d7c-4a02-8a8e-2a4cc58dfdf7\r\n*1\r\n$4\r\nEXEC\r\n", 176, 0, NULL, 0) = 176

[pengwk]

使用了那些设计模式？

• 代理模式
• 观察者模式
• 状态机模式
• 备忘录

[pengwk]

DAG directed acyclic map 有向无环图

https://zh.wikipedia.org/wiki/%E6%9C%89%E5%90%91%E6%97%A0%E7%8E%AF%E5%9B%BE

[pengwk]

default queue (named “celery” for historical reasons).

[pengwk]

The non-AMQP backends like Redis or SQS don’t support exchanges, so they require the exchange to have the same name as the queue. Using this design ensures it will work for them as well.

[pengwk]

• 前言
• celery 的基本使用
• 用例示例
• 使用工具理解从外部理解 celery
• 看代码
• 数据结构怎样，算法
• celery 的生态（监控，beat，扩展，Python 之外，竞品）

[pengwk]

Celery 的生态

管理

• Flower

监控

• Flower: Flower is a web based tool for monitoring and administrating Celery clusters
• clearly:Clear and accurate real-time monitor for celery, the distributed task queue
• Exporter

与 Web 框架结合

Django

Flask

tornado

扩展增强

RedBeat

RedBeat is a Celery Beat Scheduler that stores the scheduled tasks and runtime metadata in Redis.

Celery Once

https://github.com/cameronmaske/celery-once

Celery for node

[node-celery: Celery client for Node.js]https://github.com/mher/node-celery

node-celery-ts: TypeScript Celery client for Node

Celery for go

gocelery: Celery Distributed Task Queue in Go

竞品

[pengwk]

http://docs.celeryproject.org/en/latest/internals/guide.html#module-overview