nova-conductor与AMQP(一)

源码版本：H版

一、AMQP基础

1、什么是AMQP

可以参考如下文章：

http://blog.csdn.net/linvo/article/details/5750987

http://blog.csdn.net/gaoxingnengjisuan/article/details/11468061

2、 nova中的AMQP

可以参考如下文章：

http://docs.openstack.org/developer/nova/devref/rpc.html

3、 Qpid操作（RabbitMQ也有相应的命令查看消息队列情况）

1）安装qpid工具：

https://qpid.apache.org/download.html（安装C++ broker command-line tools 和 QMF）

2）使用qpid-tool命令查看消息队列详情，具体操作可以用help参数查看

二、以nova-conductor为例具体分析

　　凡是对外提供RPC调用的组件都会包含rpcapi.py文件，将这些调用放在该文件中。如果需要对这些组件进行rpc调用，只需导入这个rpcapi.py文件，然后创建相应的类，调用相应的函数。

　　在rpcapi.py中一般API类的函数都是调用self.client的cast函数或call函数发送消息。其中，cast函数是异步的，不需要返回结果，而call函数是同步的，需要等待结果返回。我们需要进一步了解这个self.client对象。在这些API类的构造函数中我们发现self.client对象都是通过self.get_client函数获得，由于这些API类都继承nova.rpcclient.RpcProxy类，所以其调用的get_client函数就是nova.rpcclient.RpcProxy中的get_client，该方法获得的实例为nova.rpcclient.RPCClient对象。总的来说，就是rpcapi.py中的类方法都是借用nova.rpcclient.RPCClient对象的cast或call进行消息的发送。

　　以nova.conductor.rpcapi.ComputeTaskAPI的 build_instances函数为例，这里希望通过该api调用nova-conductor为我们建立虚拟机，进一步的分析如下：

def build_instances(self, context, instances, image, filter_properties,

            admin_password, injected_files, requested_networks,

            security_groups, block_device_mapping, legacy_bdm=True):

    instances_p = [jsonutils.to_primitive(inst) for inst in instances]

    image_p = jsonutils.to_primitive(image)

    cctxt = self.client.prepare(version='1.5')

    cctxt.cast(context, 'build_instances',

               instances=instances_p, image=image_p,

               filter_properties=filter_properties,

               admin_password=admin_password,

               injected_files=injected_files,

               requested_networks=requested_networks,

               security_groups=security_groups,

               block_device_mapping=block_device_mapping,

               legacy_bdm=legacy_bdm)

　　根据上面的说明，cctxt.cast 实际为RPCClient的cast函数，如下：

def cast(self, ctxt, method, **kwargs):

    if self.server_params:

        def cast_to_server(ctxt, msg, **kwargs):

            if self.fanout:

                return self.proxy.fanout_cast_to_server(

                    ctxt, self.server_params, msg, **kwargs)

            else:

                return self.proxy.cast_to_server(

                    ctxt, self.server_params, msg, **kwargs)

        caster = cast_to_server

    else:

        """caster为ComputeTaskAPI实例的cast方法"""

        caster = self.proxy.fanout_cast if self.fanout else self.proxy.cast

    """调用caster方法"""

    self._invoke(caster, ctxt, method, **kwargs)

　　RPCClient中的self.proxy指向处理rpc的代理，这里为ComputeTaskAPI类，说了半天其实创建了RPCClient，但是RPCClient又回来调用API类自身的cast方法。由于继承关系，直接进入nova.openstack.common.rpc.proxy.py中，调用RpcProxy的cast方法。如下：

def cast(self, context, msg, topic=None, version=None):

    self._set_version(msg, version)

    msg['args'] = self._serialize_msg_args(context, msg['args'])

    """其中rpc为nova.openstack.common.rpc"""

    rpc.cast(context, self._get_topic(topic), msg)

　　进入nova/openstack/common/rpc/__init__.py模块，该模块中包含了RPC调用的接口，如下：

def cast(context, topic, msg):

    return _get_impl().cast(CONF, context, topic, msg)

　　先看_get_impl()函数：

def _get_impl():

    global _RPCIMPL

    if _RPCIMPL is None:

        try:

            _RPCIMPL = importutils.import_module(CONF.rpc_backend)

        except ImportError:

            impl = CONF.rpc_backend.replace('nova.rpc',

                                            'nova.openstack.common.rpc')

            _RPCIMPL = importutils.import_module(impl)

    return _RPCIMPL

　　此处rpc_backend为qpid，进入nova/openstack/common/rpc/impl_qpid.py。由于AMQP后端的实现有多种，所以这里对各种后端实现进行封装，对外提供统一的接口。且看cast函数：

def cast(conf, context, topic, msg):

    return rpc_amqp.cast(

        conf, context, topic, msg,

        rpc_amqp.get_connection_pool(conf, Connection))

　　该函数调用nova/openstack/common/rpc/amqp.py中的cast函数，在amqp.py中主要是抽象地利用底层的实现构建了RPC调用的请求逻辑。

　　此处， rpc_amqp.get_connection_pool(conf, Connection)可以获取AMQP连接池，这个连接池和具体的AMQP后端实现对应。接着来看rpc_amqp.cast函数，主要内容为从连接池中建立连接，然后使用impl_qpid.py中的Connection连接发送消息。后面的处理过程可以参考：http://blog.csdn.net/gaoxingnengjisuan/article/details/12230521。主要内容就是建立创建TopicPublisher并绑定exchange，然后发送消息。

　　总结流程图如下：

　　如果是使用call函数进行RPC调用的话，前面流程与cast函数类似，在此从如下代码开始分析：

def call(conf, context, topic, msg, timeout=None):

    """Sends a message on a topic and wait for a response."""

    return rpc_amqp.call(

        conf, context, topic, msg, timeout,

        rpc_amqp.get_connection_pool(conf, Connection))

　　接着是偏底层的调用：

nova/openstack/common/rpc/amqp.py

def call(conf, context, topic, msg, timeout, connection_pool):

    """Sends a message on a topic and wait for a response."""

    rv = multicall(conf, context, topic, msg, timeout, connection_pool)

    # NOTE(vish): return the last result from the multicall

    rv = list(rv)

    if not rv:

        return

    return rv[-1]

def multicall(conf, context, topic, msg, timeout, connection_pool):

    """Make a call that returns multiple times."""

    LOG.debug(_('Making synchronous call on %s ...'), topic)

    msg_id = uuid.uuid4().hex

    msg.update({'_msg_id': msg_id})

    LOG.debug(_('MSG_ID is %s') % (msg_id))

    _add_unique_id(msg)

    pack_context(msg, context)

    import dbgp

    dbgp.brk(__file__)

    with _reply_proxy_create_sem:

        if not connection_pool.reply_proxy:

            connection_pool.reply_proxy = ReplyProxy(conf, connection_pool)#1）创建处理reply消息的direct型consumer、queue和exchange

    msg.update({'_reply_q': connection_pool.reply_proxy.get_reply_q()})

    wait_msg = MulticallProxyWaiter(conf, msg_id, timeout, connection_pool)#2）创建处理reply消息的waiter(相当于结果集对象)并添加到之前的ReplyProxy对象中

    with ConnectionContext(conf, connection_pool) as conn:

        conn.topic_send(topic, rpc_common.serialize_msg(msg), timeout)#3）发送消息

    return wait_msg

　　下面用一张图简单看看call函数调用后该如何接收并返回RPC调用结果：

　　整个过程都是ReplyProxy对象在进行控制处理。其中的消息队列是direct型专门用来接收返回信息的，红色箭头指明了返回信息的数据流向，DirectConsumer监听并从消息队列中取出消息，然后将消息最后转交给MulticallProxyWaiter对象，该对象会作为结果返回，可以迭代并取出内容。注意MulticallProxyWaiter一般会在call函数调用后立即返回，此时其中并没有包含返回的消息结果，所以在迭代取出内容时会进行阻塞，直到DirectConsumer将返回消息取出并放入，此时MulticallProxyWaiter才能继续迭代。

三、额外说明

　　在消息队列的使用中，最关键的是exchange和topic，其确定了消息将发送至哪个队列，也就确定了消息将由哪个组件的哪个函数处理。所有的nova服务组件启动时都会设置exchange为nova，使用哪个组件的rpcapi.py文件中的类就会将topic初始化为该组件的topic，每个组件的topic具体内容会在/etc/nova/nova.conf中进行配置。这样exchange和topic就共同确定了消息的处理流程！！！

>>>>继续看nova-conductor与AMQP（二）

巴特西

nova-conductor与AMQP(一)

最新文章

热门文章