从本文开始，我们通过一个系列来介绍消息队列 Kombu（为后续Celery分析打基础）。

Kombu 的定位是一个兼容 AMQP 协议的消息队列抽象，是一个把消息传递封装成统一接口的库。其特点是支持多种的符合APMQ协议的消息队列系统。不仅支持原生的AMQP消息队列如RabbitMQ、Qpid，还支持虚拟的消息队列如redis、mongodb、beantalk、couchdb、in-memory等。

通过本系列，大家可以了解 Kombu 是如何实现 AMQP。本文先介绍相关概念和整体逻辑架构。

介绍 AMQP 是因为 Kombu 的定位是一个兼容 AMQP 协议的消息队列抽象。

AMQP（Advanced Message Queuing Protocol，高级消息队列协议）是一个进程间传递异步消息的网络协议。

AMQP的基本概念如下：

工作过程是：

基本如下图：

Kombu 利用了 Poll 模型，所以我们有必要介绍下。这就是IO多路复用。

IO多路复用是指内核一旦发现进程指定的一个或者多个IO条件准备读取，它就通知该进程。IO多路复用适用比如当客户处理多个描述字时（一般是交互式输入和网络套接口）。

与多进程和多线程技术相比，I/O多路复用技术的最大优势是系统开销小，系统不必创建进程/线程，也不必维护这些进程/线程，从而大大减小了系统的开销。

select 通过一个select()系统调用来监视多个文件描述符的数组（在linux中一切事物皆文件，块设备，socket连接等）。

当select()返回后，该数组中就绪的文件描述符便会被内核修改标志位（变成ready），使得进程可以获得这些文件描述符从而进行后续的读写操作（select会不断监视网络接口的某个目录下有多少文件描述符变成ready状态【在网络接口中，过来一个连接就会建立一个'文件'】，变成ready状态后，select就可以操作这个文件描述符了）。

poll 和select在本质上没有多大差别，但是poll没有最大文件描述符数量的限制。

poll和select同样存在一个缺点就是，包含大量文件描述符的数组被整体复制于用户态和内核的地址空间之间，而不论这些文件描述符是否就绪，它的开销随着文件描述符数量的增加而线性增大。

select()和poll()将就绪的文件描述符告诉进程后，如果进程没有对其进行IO操作，那么下次调用select()和poll() 的时候将再次报告这些文件描述符，所以它们一般不会丢失就绪的消息，这种方式称为水平触发（Level Triggered）。

epoll由内核直接支持，可以同时支持水平触发和边缘触发（Edge Triggered，只告诉进程哪些文件描述符刚刚变为就绪状态，它只说一遍，如果我们没有采取行动，那么它将不会再次告知，这种方式称为边缘触发），理论上边缘触发的性能要更高一些。

epoll同样只告知那些就绪的文件描述符，而且当我们调用epoll_wait()获得就绪文件描述符时，返回的不是实际的描述符，而是一个代表 就绪描述符数量的值，你只需要去epoll指定的一个数组中依次取得相应数量的文件描述符即可，这里也使用了内存映射（mmap）技术，这样便彻底省掉了 这些文件描述符在系统调用时复制的开销。

另一个本质的改进在于epoll采用基于事件的就绪通知方式。在select/poll中，进程只有在调用一定的方法后，内核才对所有监视的文件描 述符进行扫描，而epoll事先通过epoll_ctl()来注册一个文件描述符，一旦基于某个文件描述符就绪时，内核会采用类似callback的回调 机制，迅速激活这个文件描述符，当进程调用epoll_wait()时便得到通知。

阻塞I/O模式下，内核对于I/O事件的处理是阻塞或者唤醒，一个线程只能处理一个流的I/O事件。如果想要同时处理多个流，要么多进程(fork)，要么多线程(pthread_create)，很不幸这两种方法效率都不高。

非阻塞忙轮询的I/O方式可以同时处理多个流。我们只要不停的把所有流从头到尾问一遍，又从头开始。这样就可以处理多个流了，但这样的做法显然不好，因为如果所有的流都没有数据，那么只会白白浪费CPU。

非阻塞模式下可以把I/O事件交给其他对象（select以及epoll）处理甚至直接忽略。

为了避免CPU空转，可以引进一个代理（一开始有一位叫做select的代理，后来又有一位叫做poll的代理，不过两者的本质是一样的）。这个代理比较厉害，可以同时观察许多流的I/O事件，在空闲的时候，会把当前线程阻塞掉，当有一个或多个流有I/O事件时，就从阻塞态中醒来，于是我们的程序就会轮询一遍所有的流（于是我们可以把“忙”字去掉了）。代码长这样:

于是，如果没有I/O事件产生，我们的程序就会阻塞在select处。但是依然有个问题，我们从select那里仅仅知道了，有I/O事件发生了，但却并不知道是那几个流（可能有一个，多个，甚至全部），我们只能无差别轮询所有流，找出能读出数据，或者写入数据的流，对他们进行操作。

epoll可以理解为event poll，不同于忙轮询和无差别轮询，epoll只会把哪个流发生了怎样的I/O事件通知我们。此时我们对这些流的操作都是有意义的（复杂度降低到了O(1)）。

epoll版服务器实现原理类似于select版服务器，都是通过某种方式对套接字进行检验其是否能收发数据等。但是epoll版的效率要更高，同时没有上限。

在select、poll中的检验，是一种被动的轮询检验，而epoll中的检验是一种主动地事件通知检测，即：当有套接字符合检验的要求，便会主动通知，从而进行操作。这样的机制自然效率会高一点。

同时在epoll中要用到文件描述符，所谓文件描述符实质上是数字。

epoll的主要用处在于：

如果进程在处理while循环中的代码时，一些套接字对应的客户端如果发来了数据，那么操作系统底层会自动的把这些套接字对应的文件描述符写入该列表中，当进程再次执行到epoll时，就会得到了这个列表，此时这个列表中的信息就表示着哪些套接字可以进行收发了。因为epoll没有去依次的查看，而是直接拿走已经可以收发的fd，所以效率高！

Kombu的最初的实现叫做carrot，后来经过重构才成了Kombu。

Kombu 主要用途如下：

在 Kombu 中，存在多个概念（部分和AMQP类似），他们分别是：

Message：消息，发送和消费的主体，生产消费的基本单位，其实就是我们所谓的一条条消息；

Connection：对 MQ 连接的抽象，一个 Connection 就对应一个 MQ 的连接；Connection 是 AMQP 对 连接的封装；

Channel：与AMQP中概念类似，可以理解成共享一个Connection的多个轻量化连接；Channel 是 AMQP 对 MQ 的操作的封装；

Transport：kombu 支持将不同的消息中间件以插件的方式进行灵活配置，使用transport这个术语来表示一个具体的消息中间件，可以认为是对broker的抽象：

Producers: 发送消息的抽象类；

Consumers：接受消息的抽象类。consumer需要声明一个queue，并将queue与指定的exchange绑定，然后从queue里面接收消息；

Exchange：MQ 路由，这个和 RabbitMQ 差不多，支持 5 类型。消息发送者将消息发至Exchange，Exchange负责将消息分发至队列。用于路由消息（消息发给exchange，exchange发给对应的queue）。路由就是比较routing-key（这个message提供）和binding-key（这个queue注册到exchange的时候提供）。使用时，需要指定exchange的名称和类型（direct，topic和fanout）。

交换机通过匹配消息的 routing_key 和 binding_key来转发消息，binding_key 是consumer 声明队列时与交换机的绑定关系。

Queue：对应的 queue 抽象，存储着即将被应用消费掉的消息，Exchange负责将消息分发Queue，消费者从Queue接收消息。

Routing keys: 每个消息在发送时都会声明一个routing_key。routing_key的含义依赖于exchange的类型。一般说来，在AMQP标准里定义了四种默认的exchange类型，此外，vendor还可以自定义exchange的类型。最常用的三类exchange为：

以 redis 为 broker，我们简要说明：

Connection是对 MQ 连接的抽象，一个 Connection 就对应一个 MQ 的连接。现在就是对 'redis://localhost:6379' 连接进行抽象。

由之前论述可知，Connection是到broker的连接。从具体代码可以看出，Connection更接近是一个逻辑概念，具体功能都委托给别人完成。

Connection主要成员变量是：

精简版定义如下：

Channel：与AMQP中概念类似，可以理解成共享一个Connection的多个轻量化连接。就是真正的连接。

Channel 可以认为是 redis 操作和连接的封装。每个 Channel 都可以与 redis 建立一个连接，在此连接之上对 redis 进行操作，每个连接都有一个 socket，每个 socket 都有一个 file，从这个 file 可以进行 poll。

简化版定义如下：

手机如图：

Transport：真实的 MQ 连接，也是真正连接到 MQ(redis/rabbitmq) 的实例。就是存储和发送消息的实体，用来区分底层消息队列是用amqp、Redis还是其它实现的。

我们顺着上文理一下：

在Kombu 体系中，用 transport 对所有的 broker 进行了抽象，为不同的 broker 提供了一致的解决方案。通过Kombu，开发者可以根据实际需求灵活的选择或更换broker。

Transport负责具体操作，但是 很多操作移交给 loop 与 MultiChannelPoller 进行。

其主要成员变量为：

其中重点是MultiChannelPoller。一个Connection有一个Transport， 一个Transport有一个MultiChannelPoller，对poll操作都是由MultiChannelPoller完成，redis操作由channel完成。

定义如下：

MultiChannelPoller 定义如下，可以理解为 执行 engine，主要作用是：

或者从逻辑上这么理解，MultiChannelPoller 就是：

具体定义如下：

Consumer 是消息接收者。Consumer & 相关组件 的作用主要如下：

在具体 Consumer 的实现中，它把 queue 与 channel 联系起来。queue 里面有一个 channel，用来访问redis，也有 Exchange，知道访问具体 redis 哪个key（就是queue对应的那个key）。

Consumer 消费消息是通过 Queue 来消费，然后 Queue 又转嫁给 Channel。

所以服务端的逻辑大致为：

Consumer 定义如下：

此时总体逻辑如下图：

手机如下：

现在我们知道：

Producer 是消息发送者。Producer中，主要变量是：

逻辑如图：

手机如图：

用户可以通过同步方式自行读取消息，如果不想自行读取，也可以通过Hub（本身构建了一个异步消息引擎）读取。

Hub 是一个eventloop，拥有自己的 poller。

前面在 MultiChannelPoller 中间提到了，MultiChannelPoller 会建立了自己内部的 poller。但是实际上在注册时候，Transport 会使用 hub 的 poller，而非 MultiChannelPoller 内部的 poller。

Connection注册到Hub，一个Connection对应一个Hub。

在注册过程中，Hub 把自己内部的 poller 配置在 Transport 之中。这样就通过 transport 内部的 MultiChannelPoller 可以把 Hub . poller 和 Channel 对应的 socket 同poll联系起来，一个 socket 在 linux 系统中就是一个file，就可以进行 poll 操作；

因而，如前面所述，这样 MultiChannelPoller 就可以 打通 Channel ---> socket ---> poll ---> fd ---> 读取 redis 这条通路了，就是如果 redis 有数据来了，MultiChannelPoller 就马上通过 poll 得到通知，就去 redis 读取。

Hub定义如下：

我们通过文字和图例来总结下本文。

Message：消息，发送和消费的主体，其实就是我们所谓的一条条消息；

Connection是AMQP对消息队列连接的封装抽象，那么两者的关系就是：对MQ的操作必然离不开连接。

Channel是AMQP对MQ的操作的封装，可以理解成共享一个Connection的多个轻量化连接。

Kombu并不直接让Channel使用Connection来发送/接受请求，而是引入了一个新的抽象Transport，Transport负责具体的MQ的操作，也就是说Channel的操作都会落到Transport上执行。是以Transport为中心，把Channel代表的真实redis与Hub其中的poll联系起来。

Queue：消息队列，消息内容的载体，存储着即将被应用消费掉的消息。Exchange 负责将消息分发 Queue，消费者从 Queue 接收消息；

Exchange：交换机，消息发送者将消息发至 Exchange，Exchange 负责将消息分发至 Queue；

Producers: 发送消息的抽象类，Producer 包含了很多东西，有 Exchange、routing_key 和 channel 等等；

Consumers：接受消息的抽象类，consumer需要声明一个queue，并将queue与指定的exchange绑定，然后从queue里面接收消息；

用户可以通过同步方式自行读取消息，如果不想自行读取，也可以通过Hub（本身构建了一个异步消息引擎）读取。

Hub是一个eventloop，Connection注册到Hub，一个Connection对应一个Hub。Hub 把自己内部的 poller 配置在 Transport 之中。这样就通过 transport 内部的 MultiChannelPoller 可以把 Hub . poller 和 Channel 对应的 socket 同poll联系起来，一个 socket 在 linux 系统中就是一个file，就可以进行 poll 操作；

MultiChannelPoller是Connection 和 Hub的枢纽，它负责找出哪个 Channel 是可用的，但是这些 Channel 都是来自同一个 Connection。

具体如图，可以看到，