Celery解析系列


玩 Python 的同学可能很多都听说过甚至玩过 Celery,Celery 作为 Python 中最流行的异步消息队列可以说是非常得受欢迎。但是,用得多的同学相信也是有一个感触,那就是时不时会遇到莫名奇妙的坑,而且最后都只能通过重启解决,因为要找到原因实在不是一件容易的事情。

而我作为这受伤人群中的一个,想了解一下这个大坑内部的原理,所以就曾经扒过它内部的实现,但是,由于 Celery 确实有点庞大,而且我认为代码实现得也不是很多,所以,只是学习到了一部分的知识,但是,却是收获良多。所以,最近因为各种情况,我这次深扒一次,所以后续将会有一系列的文章是讲 Celery 实现的。

当你尝试查看 Celery 的源码的时候,你会发现有一个你根本绕不过的坑的,那就是 Kombu,Kombu 的定位是一个兼容 AMQP 协议的消息队列抽象,但是,这只是作者的一面说辞,你看完代码就知道了,它是尝试做异步MQ 的兼容 AMQP 的抽象队列。所以,我决定先从 Kombu 的代码写起,而这篇文章是第一篇!

我在这个系列里面讲的都是 Kombu 4.1.0 和 Celery 4.1.0 的事,先提前说一下,以免后续的同学对不上号。

下载 Kombu 代码

Kombu 的代码很好找,因为它和 Celery 是强耦合的,所以,它的代码放在 Celery 的代码组里头,可以很简单得 Github 上找到,所以下载也是很简单了:

> git clone https://github.com/celery/kombu.git
> cd kombu
> git checkout v4.1.0

这样,你就得到了我这个系列文章里面的 Kombu 的源代码了,而我后面的所有代码来源以及行数都是对应这个版本的代码的!

整体 Review

拿到代码之后,我就先摒弃所有的非代码部分,直接来整体看看 kombu 这个目录下有什么:

可以看到,在 kombu 这个目录里面大部分都是文件,而只有三个文件夹,分别是:

从这个目录中,我们就发现了一些问题,你说你既然是抽象 MQ 的,你要啥异步操作啊,这有点越界了呀喂。

简单添加/获取消息

OK,概览完我们看一些实际的东西,先尝试用 Kombu 写一个生产/消费的代码吧,这个 DEMO 分为两部分

下面看看这两段代码都是怎么写的:

还是老规矩,这两段代码你都可以在我的 Github 中找到。先看看看消费者,消费者的代码比较简单,我们先建立和 MQ 的连接,然后再从 MQ 的指定队列里面将消息拿出来,处理掉。

而生产者则相反,前面也是要先和队列建立连接,但是,有一点不一样的是这里创建了一个 exchange,然后再往队列里面发送消息,而发送消息的同时还是搭配好多参数。

这里就是 Kombu 有意思的一点了,它意图对所有的 MQ 进行抽象,然后通过接口对外暴露出一致的 API,这样我们就不用关心底层用的是什么 MQ 了,Redis/RabbitMQ/MongoDB 之类的随便切换。

Kombu 的 MQ 模型

因为 Kombu 是对 AMQP 进行抽象,所以它必定有抽象的模型,事实上,它大体上和 RabbitMQ 差不多,但是,不完全一样,有一些差别,下面就介绍一下 Konbu 的抽象模型。

在 Kombu 中,存在多个概念,其实我们在前边简单的生产/消费者样例中已经看到了了一些,他们分别是:

消息是发送到那个 Queue 的

假设我们想要发送一个消息到 Redis 中名为 ‘test’ 的 queue 中,那么 Kombu 是怎么做的,这就设计到 Exchange 的概念了。目前 Kombu 对于不同 MQ 的支持是这样的:

假设我们用的是 Direct,那么我们的 Producer 在生产的时候只需要指定 Queue=test 即可,这样就会发送的 test 这个 queue 中。更多关于 Exchange 的知识可以参考:AMQP 0-9-1 Model Explained

这一篇就先介绍到这里,后续就开始深入到代码层面,看下这个模型中的各个实体是如何实现的。

Reference

  1. AMQP 0-9-1 Model Explained
  2. Kombu Docs