感謝朋友支援本部落格，歡迎共同探討交流，由於能力和時間有限，錯誤之處在所難免，歡迎指正！
如果轉載，請保留作者資訊。
部落格地址：http://blog.csdn.net/gaoxingnengjisuan
郵箱地址：[email protected]

這篇博文開始解析NOVA中的AMQP架構下訊息的消費者如何從特定的訊息佇列中讀取傳送給自己的訊息，並進行執行操作。

總體來講，Nova中的各個服務在啟動的時候就會初始化會用到的佇列，而且會啟動一個綠色執行緒，不斷的迴圈驗證新的訊息的到來，一旦有新的訊息，將會由合適的consumer進行讀取，並進一步進行訊息的解析和執行操作。

下面，我將會以compute服務的啟動作為例項，重點解析AMQP架構下訊息的消費操作。

1.kombu consumer程式碼示例

kombu中訊息接收消費機制的簡單實現如示例所示：

    #!/usr/bin/python  
      
    from kombu.entity import Exchange, Queue  
    from kombu.messaging import Consumer  
    from kombu.connection import Connection  
      
      
      
    def process_media(body, message):  
        print body  
        message.ack()  
      
    connection = Connection('amqp://guest:
 
[email protected]:5672//')  
    channel = connection.channel()  
      
    media_exchange = Exchange('media', 'direct', channel)  
    video_queue = Queue('video', exchange=media_exchange, routing_key='video', channel=channel)  
      
      
    consumer =  Consumer(channel, queues=[video_queue], callbacks=[process_media])   
    consumer.consume()  
      
    while True:  
        connection.drain_events()  
    consumer.cancel()  
 

（1）建立連線；

（2）獲取channel；

（3）建立exchange；

（4）建立佇列並與exchange繫結；

（5）建立Consumer；

（6）consume()向server註冊，表明可以接受訊息了；

（7）drain_enents阻塞程式，等待訊息到來；

（8）cancel（）通知server不要向該consumer傳送任何訊息了；

在nova中，當然不能實現的這麼簡便，而是進行了一系列的封裝操作，但是基本的實現思路是一致的。

2.以nova-compute服務啟動為例，解析AMQP架構下訊息的消費操作

首先來看nova-compute服務啟動指令碼程式碼：

"""
Nova Compute的啟動指令碼；
"""

import eventlet
import os

if os.name == 'nt':
    eventlet.monkey_patch(os=False)
else:
    eventlet.monkey_patch()

import os
import sys
import traceback

from oslo.config import cfg

POSSIBLE_TOPDIR = os.path.normpath(os.path.join(os.path.abspath(sys.argv[0]),
                                   os.pardir,
                                   os.pardir))
if os.path.exists(os.path.join(POSSIBLE_TOPDIR, 'nova', '__init__.py')):
    sys.path.insert(0, POSSIBLE_TOPDIR)


from nova import config
import nova.db.api
from nova import exception
from nova.openstack.common import log as logging
from nova import service
from nova import utils

CONF = cfg.CONF
CONF.import_opt('compute_topic', 'nova.compute.rpcapi')
CONF.import_opt('use_local', 'nova.conductor.api', group='conductor')
LOG = logging.getLogger('nova.compute')


def block_db_access():
    class NoDB(object):
        def __getattr__(self, attr):
            return self

        def __call__(self, *args, **kwargs):
            stacktrace = "".join(traceback.format_stack())
            LOG.error('No db access allowed in nova-compute: %s' % stacktrace)
            raise exception.DBNotAllowed('nova-compute')

    nova.db.api.IMPL = NoDB()


if __name__ == '__main__':
    config.parse_args(sys.argv)
    logging.setup('nova')
    utils.monkey_patch()

    if not CONF.conductor.use_local:
        block_db_access()

    # 初始化Service這個類，並且獲取其例項化物件；
    server = service.Service.create(binary='nova-compute',
                                    topic=CONF.compute_topic,
                                    db_allowed=False)
    service.serve(server)
    service.wait()

def serve(server, workers=None):
    global _launcher
    if _launcher:
        raise RuntimeError(_('serve() can only be called once'))

    if workers:
        _launcher = ProcessLauncher()
        _launcher.launch_server(server, workers=workers)
    else:
        _launcher = ServiceLauncher()
        _launcher.launch_server(server)

def launch_server(self, server):
    """
    載入並啟動給定的服務；
    """
    if self.backdoor_port is not None:
        server.backdoor_port = self.backdoor_port
    gt = eventlet.spawn(self.run_server, server)
    self._services.append(gt)

def run_server(server):
    """
    啟動並等待一個服務的執行的結束；
    """
    server.start()
    server.wait()

    def start(self):
        verstr = version.version_string_with_package()
        LOG.audit(_('Starting %(topic)s node (version %(version)s)'),
                  {'topic': self.topic, 'version': verstr})
        # 在程序開始之前執行基本配置的檢測；
        self.basic_config_check()
        self.manager.init_host()
        self.model_disconnected = False
        # 獲取上下文資訊；
        ctxt = context.get_admin_context()
        try:
            # 查詢資料庫獲取topic、host、binary型別指定的所有的服務；
            self.service_ref = self.conductor_api.service_get_by_args(ctxt, self.host, self.binary)
            # 獲取這些服務的ID值；
            self.service_id = self.service_ref['id']
        except exception.NotFound:
            # 如果沒有合適的服務存在，則根據上下文要求建立一個服務環境；
            self.service_ref = self._create_service_ref(ctxt)

        if self.backdoor_port is not None:
            self.manager.backdoor_port = self.backdoor_port

        # 建立一個到用於RPC的訊息匯流排的連線；
        # 建立獲取到RabbitMQ的連線；
        # 建立連線，預設是kombu實現；
        self.conn = rpc.create_connection(new=True)
        LOG.debug(_("Creating Consumer connection for Service %s") %
                  self.topic)
        # 更新現有資源；
        # 讀取系統的總共資源以及可用的資源，更新資源，算出已經使用的資源；
        self.manager.pre_start_hook(rpc_connection=self.conn)

        # 獲取RPC排程器；
        # 初始化RPC排程器；
        rpc_dispatcher = self.manager.create_rpc_dispatcher()

        # 建立不同的訊息消費者；
        # 建立以服務的topic為路由鍵的消費者；
        self.conn.create_consumer(self.topic, rpc_dispatcher, fanout=False)

        # 建立以服務的topic和本機名為路由鍵的消費者（基於topic&host，可用來接收定向訊息）;
        node_topic = '%s.%s' % (self.topic, self.host)
        self.conn.create_consumer(node_topic, rpc_dispatcher, fanout=False)

        # fanout直接投遞訊息，不進行匹配，速度最快（fanout型別，可用於接收廣播訊息）；
        self.conn.create_consumer(self.topic, rpc_dispatcher, fanout=True)

        # 啟動消費者執行緒；
        # consume_in_thread用evelent.spawn建立一個協程一直執行；
        # 等待訊息，在有消費到來時會建立新的協程執行遠端呼叫的函式；
        self.conn.consume_in_thread()

        # 在啟動服務之後，並能夠通過RPC接收訊息之後，廣播通知每個節點更新capabilities屬性值，由控制器獲取每個節點上的這個屬性值；
        self.manager.post_start_hook()

        LOG.debug(_("Join ServiceGroup membership for this service %s")
                  % self.topic)
 # 新增服務到服務成員組；
        pulse = self.servicegroup_api.join(self.host, self.topic, self)
        if pulse:
            self.timers.append(pulse)

        if self.periodic_enable:
            if self.periodic_fuzzy_delay:
                initial_delay = random.randint(0, self.periodic_fuzzy_delay)
            else:
                initial_delay = None

            periodic = utils.DynamicLoopingCall(self.periodic_tasks)
            periodic.start(initial_delay=initial_delay,
                           periodic_interval_max=self.periodic_interval_max)
            self.timers.append(periodic)

2.1 語句self.conn = rpc.create_connection(new=True)的解析

這條語句實現了建立獲取到RabbitMQ的連線，具體來看方法create_connection的程式碼實現：

def create_connection(new=True):
    """
    建立一個到用於RPC的訊息匯流排的連線；
    建立獲取到RabbitMQ的連線；
    """
    return _get_impl().create_connection(CONF, new=new)

def create_connection(conf, new=True):
    """
    建立獲取到RabbitMQ的連線；
    """
    
    # get_connection_pool：獲取到RabbitMQ的連線池，並返回這個連線池的物件；
    # Connection：連線到RabbitMQ的實現類；
    return rpc_amqp.create_connection(
        conf, new,
        rpc_amqp.get_connection_pool(conf, Connection))

def create_connection(conf, new, connection_pool):
    """
    建立連線；
    """
    return ConnectionContext(conf, connection_pool, pooled=not new)

class ConnectionContext(rpc_common.Connection):
    """
    這個類是對連線功能的一個封裝；
    這個類提供方法建立新的連線，也可以實現從連線池中直接獲取一個連線；
    當然，也有方法可以實現對連線的刪除，當連線刪除之後，如果是從連線池獲取的連線會把連線返回連線池；
    """

    def __init__(self, conf, connection_pool, pooled=True, server_params=None):
        """
        建立一個新的連線，或者從連線池中獲取一個連線；
        """
        self.connection = None
        self.conf = conf
        self.connection_pool = connection_pool
        
        # 如果已經獲取連線池物件，直接從連線池中獲取一個連線；
        if pooled:
            self.connection = connection_pool.get()
        else:
            self.connection = connection_pool.connection_cls(conf, server_params=server_params)
        self.pooled = pooled

2.2 語句self.conn.create_consumer(self.topic, rpc_dispatcher, fanout=False)的解析

這裡有三條相似的語句，如下：

    # 建立不同的訊息消費者；
    # 建立以服務的topic為路由鍵的消費者；
    self.conn.create_consumer(self.topic, rpc_dispatcher, fanout=False)

    # 建立以服務的topic和本機名為路由鍵的消費者（基於topic&host，可用來接收定向訊息）;
    node_topic = '%s.%s' % (self.topic, self.host)
    self.conn.create_consumer(node_topic, rpc_dispatcher, fanout=False)

    # fanout直接投遞訊息，不進行匹配，速度最快（fanout型別，可用於接收廣播訊息）；
    self.conn.create_consumer(self.topic, rpc_dispatcher, fanout=True)

這裡現以語句self.conn.create_consumer(self.topic, rpc_dispatcher, fanout=False)作為示例進行解析。

首先來看方法create_consumer，實現了訊息消費者的建立，具體來看程式碼的實現：

def create_consumer(self, topic, proxy, fanout=False):
        """
        根據引數建立具體的訊息消費者；
        """
        
        # Connection：連線到RabbitMQ的實現類；
        # get_connection_pool：獲取到RabbitMQ的連線池，並返回這個連線池的物件；
        # connection_cls：是一個連線類的物件；
        # 獲取ProxyCallback類的初始化例項物件；
        proxy_cb = rpc_amqp.ProxyCallback(
            self.conf, proxy,
            rpc_amqp.get_connection_pool(self.conf, Connection))
        self.proxy_callbacks.append(proxy_cb)

        if fanout:
            # 建立一個廣播型別的消費者；
            # 宣告佇列和交換器，通過routing key繫結佇列到交換器；
            self.declare_fanout_consumer(topic, proxy_cb)
        else:
            # declare_topic_consumer：建立一個主題式的資訊消費者；
            # 宣告佇列和交換器，通過routing key繫結佇列到交換器；
            self.declare_topic_consumer(topic, proxy_cb)

class ProxyCallback(_ThreadPoolWithWait):
    def __init__(self, conf, proxy, connection_pool):
        # 這個類實現了啟動一個用於處理傳入資訊的綠色執行緒；
        super(ProxyCallback, self).__init__(
            conf=conf,
            connection_pool=connection_pool,
        )
        self.proxy = proxy
        self.msg_id_cache = _MsgIdCache()

    def __call__(self, message_data):
        if hasattr(local.store, 'context'):
            del local.store.context
        # 記錄日誌，但是一些敏感資訊，像建立虛擬機器時的密碼等資訊不會
        # 顯示在日誌中，面是替換為"<SANITIZED>"
        rpc_common._safe_log(LOG.debug, _('received %s'), message_data)
        # 重複訊息檢測；
        # ack返回前，AMQP消費者可能會出現兩次讀取相同資訊的異常，這個方法可以防止這樣的情況出現；
        self.msg_id_cache.check_duplicate_message(message_data)
        # 從message_data中解析出上下文資訊；
        ctxt = unpack_context(self.conf, message_data)
        # 從message_data中獲取要執行的方法method；
        method = message_data.get('method')
        # 從message_data中獲取相關引數args；
        args = message_data.get('args', {})
        # 從message_data中獲取版本的相關資訊；
        version = message_data.get('version', None)
        if not method:
            LOG.warn(_('no method for message: %s') % message_data)
            ctxt.reply(_('No method for message: %s') % message_data,
                       connection_pool=self.connection_pool)
            return
        # 建立一個新的綠色執行緒來執行方法method；
        self.pool.spawn_n(self._process_data, ctxt, version, method, args)

class _ThreadPoolWithWait(object):
    """
    這個類實現了啟動一個用於處理傳入資訊的綠色執行緒；
    """

    def __init__(self, conf, connection_pool):
        self.pool = greenpool.GreenPool(conf.rpc_thread_pool_size)
        self.connection_pool = connection_pool
        self.conf = conf

這裡的_call_方法是非常非常重要的，當然這裡我只是拿出來給大家看一下，這裡還沒有呼叫這個方法，這個方法就是執行了對獲取的訊息進行最終處理的過程。其實nova中呼叫這個方法真的是輾轉挪移，七拐八拐，不留意的話很難看到這個方法的呼叫，後面我會專門進行解析說明這個方法是怎麼被呼叫的。

2.2.2 語句self.declare_topic_consumer(topic, proxy_cb)的解析

這條語句實現了建立一個主題式的訊息消費者，具體來看方法declare_topic_consumer的程式碼：

def declare_topic_consumer(self, topic, callback=None, queue_name=None,
                               exchange_name=None):
    """
    建立一個主題式的資訊消費者；
    宣告佇列和交換器，通過routing key繫結佇列到交換器；
    """
        
    # 根據傳入的類建立資訊消費者，把建立好的消費者物件加入到consumers列表中；
    self.declare_consumer(functools.partial(TopicConsumer,
                                            name=queue_name,
                                            exchange_name=exchange_name,
                                            ),
                          topic, callback)

def declare_consumer(self, consumer_cls, topic, callback):
        """
        根據傳入的類建立資訊消費者，把建立好的消費者物件加入到consumers列表中；
        """

        def _connect_error(exc):
            log_info = {'topic': topic, 'err_str': str(exc)}
            LOG.error(_("Failed to declare consumer for topic '%(topic)s': "
                      "%(err_str)s") % log_info)

        def _declare_consumer():
            # 傳遞進來的類consumer_cls應該是TopicConsumer；
            # 宣告佇列和交換器，通過routing key繫結佇列到交換器；
            consumer = consumer_cls(self.conf, self.channel, topic, callback,
                                    self.consumer_num.next())
            self.consumers.append(consumer)
            return consumer

        return self.ensure(_connect_error, _declare_consumer)

下面對類TopicConsumer的初始化過程進行比較細緻的解析，來看程式碼的實現：

class TopicConsumer(ConsumerBase):
    """
    主題式訊息消費者類；
    """

    def __init__(self, conf, channel, topic, callback, tag, name=None,
                 exchange_name=None, **kwargs):
        # 預設選項的設定；
        
        # rabbit_durable_queues：這個引數定義了在RabbitMQ中是否使用永續性的佇列；
        # 引數的預設值為False；
        options = {'durable': conf.rabbit_durable_queues,
                   'queue_arguments': _get_queue_arguments(conf),
                   'auto_delete': False,
                   'exclusive': False}
        options.update(kwargs)
        
        # 獲取交換器的名稱；
        exchange_name = exchange_name or rpc_amqp.get_control_exchange(conf)
        # 根據相關的配置引數宣告產生一個交換器；
        exchange = kombu.entity.Exchange(name=exchange_name,
                                         type='topic',
                                         durable=options['durable'],
                                         auto_delete=options['auto_delete'])
        # 重連線到channel；
        # 宣告佇列和交換器，通過routing key繫結佇列到交換器；
        super(TopicConsumer, self).__init__(channel,
                                            callback,
                                            tag,
                                            name=name or topic,
                                            exchange=exchange,
                                            routing_key=topic,
                                            **options)

class Exchange(MaybeChannelBound):
    """
    交換器宣告；
    """
    # 短暫性訊息傳遞方式；
    TRANSIENT_DELIVERY_MODE = TRANSIENT_DELIVERY_MODE
    # 永續性訊息傳遞方式；
    PERSISTENT_DELIVERY_MODE = PERSISTENT_DELIVERY_MODE

    # 交換器名稱；
    name = ""
    # 如果沒有定義交換器型別，預設的交換器型別（直接式交換器），但是這裡面已經定義為主題式交換器；
    type = "direct"
    # 預設交換器是持久型別的；
    durable = True
    auto_delete = False
    # 永續性訊息傳遞方式；
    delivery_mode = PERSISTENT_DELIVERY_MODE

    attrs = (("name", None),
             ("type", None),
             ("arguments", None),
             ("durable", bool),
             ("auto_delete", bool),
             ("delivery_mode", lambda m: DELIVERY_MODES.get(m) or m))

    def __init__(self, name="", type="", channel=None, **kwargs):
        super(Exchange, self).__init__(**kwargs)
        self.name = name or self.name
        self.type = type or self.type
        # maybe_bind：如果沒有繫結，則繫結例項到channel；
        self.maybe_bind(channel)

def maybe_bind(self, channel):
        """
        如果沒有繫結，則繫結例項到channel；
        """
        if not self.is_bound and channel:
            self._channel = channel
            self.when_bound()
            self._is_bound = True
        return self

class ConsumerBase(object):
    """
    Consumer（消費者）基類；
    """

    def __init__(self, channel, callback, tag, **kwargs):
        """
        根據相關引數在一個amqp channel上宣告產生一個佇列；
        """
        self.callback = callback
        self.tag = str(tag)
        self.kwargs = kwargs
        self.queue = None
        # 重連線到channel；
        # 宣告佇列和交換器，通過routing key繫結佇列到交換器；
        self.reconnect(channel)

def reconnect(self, channel):
        """
        重連線到channel；
        宣告佇列和交換器，通過routing key繫結佇列到交換器；
        """
        self.channel = channel
        self.kwargs['channel'] = channel
        # 佇列的宣告類，實現佇列初始化，並且實現佇列與交換器和channel的繫結；
        self.queue = kombu.entity.Queue(**self.kwargs)
        
        # 序列重新的申報；
        # 宣告佇列和交換器，通過routing key繫結佇列到交換器；
        self.queue.declare()

class Queue(MaybeChannelBound):
    """
    佇列的宣告類，實現佇列初始化，並且實現佇列與交換器和channel的繫結；
    """
    name = ""
    exchange = None
    routing_key = ""

    durable = True
    exclusive = False
    auto_delete = False
    no_ack = False

    attrs = (("name", None),
             ("exchange", None),
             ("routing_key", None),
             ("queue_arguments", None),
             ("binding_arguments", None),
             ("durable", bool),
             ("exclusive", bool),
             ("auto_delete", bool),
             ("no_ack", None),
             ("alias", None))

    def __init__(self, name="", exchange=None, routing_key="", channel=None,
            **kwargs):
        super(Queue, self).__init__(**kwargs)
        self.name = name or self.name
        self.exchange = exchange or self.exchange
        self.routing_key = routing_key or self.routing_key
        # exclusive implies auto-delete.
        if self.exclusive:
            self.auto_delete = True
        # 如果沒有繫結，則繫結到channel；
        self.maybe_bind(channel)

再來看語句self.queue.declare()，這條語句實現了宣告佇列和交換器，通過routing key繫結佇列到交換器，具體的實現來看方法declare的程式碼：

def declare(self, nowait=False):
    """
    宣告佇列和交換器，通過routing key繫結佇列到交換器；
    """
    # 佇列名稱；
    name = self.name
    if name:
        if self.exchange:
            self.exchange.declare(nowait)
    self.queue_declare(nowait, passive=False)
        
    # 通過routing key繫結佇列到交換器；
    if name:
        self.queue_bind(nowait)
    return self.name

首先來看語句self.exchange.declare(nowait)，這條語句實現了對特定的交換器的宣告操作，具體來看方法declare的程式碼實現：

def declare(self, nowait=False):
        """
        宣告交換器，並在代理上建立交換器；
        """
        
        # 在綠色執行緒中執行方法exchange_declare，實現交換器的宣告和繫結；
        return _SYN(self.channel.exchange_declare, exchange=self.name,
                                                type=self.type,
                                                durable=self.durable,
                                                auto_delete=self.auto_delete,
                                                arguments=self.arguments,
                                                nowait=nowait)

def exchange_declare(self, exchange, type="direct", durable=False,
            auto_delete=False, arguments=None, nowait=False):
        """
        宣告交換器；
        """
        try:
            prev = self.state.exchanges[exchange]
            if not self.typeof(exchange).equivalent(prev, exchange, type,
                                                    durable, auto_delete,
                                                    arguments):
                raise NotEquivalentError(
                        "Cannot redeclare exchange %r in vhost %r with "
                        "different type, durable or autodelete value" % (
                            exchange,
                            self.connection.client.virtual_host or "/"))
        except KeyError:
            self.state.exchanges[exchange] = {
                    "type": type,
                    "durable": durable,
                    "auto_delete": auto_delete,
                    "arguments": arguments or {},
                    "table": [],
            }

def queue_declare(self, nowait=False, passive=False):
        """
        在伺服器上宣告佇列；
        """
        ret = _SYN(self.channel.queue_declare, queue=self.name,
                                               passive=passive,
                                               durable=self.durable,
                                               exclusive=self.exclusive,
                                               auto_delete=self.auto_delete,
                                               arguments=self.queue_arguments,
                                               nowait=nowait)
        if not self.name:
            self.name = ret[0]
        return ret

def queue_declare(self, queue, passive=False, auto_delete=False, **kwargs):
        """
        宣告佇列；
        """
        if auto_delete:
            self.auto_delete_queues.setdefault(queue, 0)
        if passive and not self._has_queue(queue, **kwargs):
            raise StdChannelError("404",
                    u"NOT_FOUND - no queue %r in vhost %r" % (
                        queue, self.connection.client.virtual_host or '/'),
                    (50, 10), "Channel.queue_declare")
        else:
            self._new_queue(queue, **kwargs)
        return queue, self._size(queue), 0

def queue_bind(self, nowait=False):
        """
        在伺服器上建立佇列的繫結；
        通過routing key繫結佇列到交換器；
        """
        return _SYN(self.channel.queue_bind, queue=self.name,
                                             exchange=self.exchange.name,
                                             routing_key=self.routing_key,
                                             arguments=self.binding_arguments,
                                             nowait=nowait)

def queue_bind(self, queue, exchange, routing_key, arguments=None,
            **kwargs):
        """
        通過routing key繫結佇列到交換器；
        """
        # 如果佇列已經在繫結的列表中，則直接返回；
        if queue in self.state.bindings:
            return
        table = self.state.exchanges[exchange].setdefault("table", [])
        # 佇列繫結的資訊加入到列表中；
        self.state.bindings[queue] = exchange, routing_key, arguments
        meta = self.typeof(exchange).prepare_bind(queue,
                                                  exchange,
                                                  routing_key,
                                                  arguments)
        table.append(meta)
        if self.supports_fanout:
            self._queue_bind(exchange, *meta)

def prepare_bind(self, queue, exchange, routing_key, arguments):
        return routing_key, self.key_to_pattern(routing_key), queue

在下一篇博文中，將繼續解析/nova/server.py中類Service下的方法def start(self)中的下一條也是很重要的語句：self.conn.consume_in_thread()，這條語句實現了從佇列中獲取訊息，並最終實現了對訊息的處理和執行操作。