在前面的几篇文章中,我们已经将 Kombu 的收发消息都讲完了,似乎已经没有什么内容。但是,事实上,在我们快速走读这些代码的时候,我们忽略了一些很重要的东西,其中给一个就是获取消息的时候,我们说过 Consumer 的 consume 是非阻塞的,然后就进行到了一个循环,等待一个 buffer 被填满,然后再返回这个 buffer。但是,我们却是不知道这是如何实现的,所以,这里我们就来八一八后面的实现。

Consumer

最开始我们肯定是先看看 consume 非阻塞是怎么实现的啦,经过几篇代码走读我们应该可以很简单得找到 Consumer 的代码了:kombu/messaging.py,然后看看 consume 操作:

可以看到,好像没啥特别亮点的,如果有的话,那就应该有两处:

  1. 不能取消监听的 queue,除非调用 cancel_by_queue
  2. 这里对所有的 queue 都是 no_wait=True,只留了一个设置了 no_wait=False,有什么用呢?继续看看

这里多了一个 tag 的概念,通过查看 Kombu 的文档我们可以发现:

Consumer_tag: Unique identifier for the consumer. The consumer tag is local to a connection, so two clients can use the same consumer tags. If this field is empty the server will generate a unique tag.

ok,就这么理解着先吧,然后看下真正 consume 的代码,这里传递了几个参数:

先不看具体的 callback 是啥,继续往深处看 consume,这里就设计到 queue 了,这个结构我们还没在之前看过,这里顺带一看:kombu/entity.py

代码很简单,直接交给 channel 了,已经是老朋友了,直接跟过去:

很简单的一段,直接设置一下各种值,然后关键在于最后一句:_reset_cycle,里面是啥,看看就了然了:

然后似乎就完了,调用就 over 了,啥事都没有,回家。

消息回调

等等等,似乎还有事情没完啊,这里只是将任务消费者声明出去了,但是,消息收过来之后放哪还不知道啊,怎么放进去的也不知道啊,这怎么办?那就继续看下我们的 sample 后面的代码呀:

self.channel.connection.client.drain_events(timeout=remaining)

这一句就重要啦,我们找出来看看这里的代码是啥,位置在 kombu/connection.py

又转化到 Transport 上了,那没办法咯,避不开的嘛。。。

这里不久衔接上咯,看看这里的实现,关键在于 Line 961 中的 get,其实就对应到 kombu/utils/scheduling.py

fun 就是我们前边设置得咯,实际值应该是 kombu/transport/virtual/base.py

然后就调到 kombu/transport/redis.py

到此就真的结束了,到此,我们可以认识到所谓的异步其实就是轮训,Kombu 对应所有需要监听的 queues 都查询一遍,直到查询完毕或者遇到一个可以使用的 queue,然后就获取内容,回调对应的 callback!

回顾

在这个跟踪过程,我们可以发现兜兜转转,但是历经的都是这么几个文件,它们分别是:

Reference

  1. Kombu Code