源码版本：H版
一、首先看服务的启动脚本
/usr/bin/nova-conductor
import sysfrom nova.cmd.conductor import mainif __name__ == "__main__": sys.exit(main())
nova/cmd/conductor.py
def main(): ... server = service.Service.create(binary='nova-conductor', topic=CONF.conductor.topic, manager=CONF.conductor.manager) service.serve(server, workers=CONF.conductor.workers) service.wait()
二、分析RPC服务的创建
nova/service.py
Service类：@classmethoddef create(cls, host=None, binary=None, topic=None, manager=None, report_interval=None, periodic_enable=None, periodic_fuzzy_delay=None, periodic_interval_max=None, db_allowed=True): ... service_obj = cls(host, binary, topic, manager, report_interval=report_interval, periodic_enable=periodic_enable, periodic_fuzzy_delay=periodic_fuzzy_delay, periodic_interval_max=periodic_interval_max, db_allowed=db_allowed) return service_obj
三、分析服务的运行
nova/service.py
def serve(server, workers=None): global _launcher if _launcher: raise RuntimeError(_('serve() can only be called once')) """ service为nova.openstack.common.service""" _launcher = service.launch(server, workers=workers) nova/openstack/common/service.py
def launch(service, workers=None): if workers: launcher = ProcessLauncher() launcher.launch_service(service, workers=workers) else: launcher = ServiceLauncher() launcher.launch_service(service) return launcher
参考nova-api的服务启动过程（）,即启动协程（使用eventlet）运行Service，主要是start函数。注意，如果在配置文件中指明多个workers的话，将有多个进程监听消息队列，取得消息的进程负责处理。接着看start函数的代码，如下：
nova/service.py
Service类：def start(self): ... self.manager.pre_start_hook() if self.backdoor_port is not None: self.manager.backdoor_port = self.backdoor_port """创建AMQP连接，方便后面创建Consumer""" self.conn = rpc.create_connection(new=True) LOG.debug(_("Creating Consumer connection for Service %s") % self.topic) rpc_dispatcher = self.manager.create_rpc_dispatcher(self.backdoor_port) self.conn.create_consumer(self.topic, rpc_dispatcher, fanout=False) node_topic = '%s.%s' % (self.topic, self.host) self.conn.create_consumer(node_topic, rpc_dispatcher, fanout=False) self.conn.create_consumer(self.topic, rpc_dispatcher, fanout=True) self.conn.consume_in_thread() self.manager.post_start_hook() LOG.debug(_("Join ServiceGroup membership for this service %s") % self.topic) self.servicegroup_api.join(self.host, self.topic, self) if self.periodic_enable: if self.periodic_fuzzy_delay: initial_delay = random.randint(0, self.periodic_fuzzy_delay) else: initial_delay = None self.tg.add_dynamic_timer(self.periodic_tasks, initial_delay=initial_delay, periodic_interval_max= self.periodic_interval_max) self.conn.create_consumer(node_topic, rpc_dispatcher, fanout=False)。这里的self.conn主要指nova.openstack.common.rpc.impl_qpid.Connection，（此处设AMQP后端实现为qpid，这个可以在nova.conf中进行配置）所以调用的create_consumer代码如下：
def create_consumer(self, topic, proxy, fanout=False): proxy_cb = rpc_amqp.ProxyCallback( self.conf, proxy, rpc_amqp.get_connection_pool(self.conf, Connection)) self.proxy_callbacks.append(proxy_cb) if fanout: consumer = FanoutConsumer(self.conf, self.session, topic, proxy_cb) else: consumer = TopicConsumer(self.conf, self.session, topic, proxy_cb) self._register_consumer(consumer) return consumer
该函数会根据具体的后端AMQP实现，向AMQP服务器发送队列和exchange创建(第一次会创建，后面会复用)和绑定请求，这样就可以将Consumer和具体的队列绑定并进行监听。self.conn.consume_in_thread函数就是负责启动Consumer线程，其使用evelent.spawn创建一个协程一直运行等待消息，在有消息到来时会创建新的协程运行远程调用的函数。当队列有消息到来时，调用proxy_cb进行处理，即ProxyCallback对象的__call__函数，代码如下：
nova/openstack/common/rpc/amqp.py
ProxyCallback类：def __call__(self, message_data): if hasattr(local.store, 'context'): del local.store.context rpc_common._safe_log(LOG.debug, _('received %s'), message_data) self.msg_id_cache.check_duplicate_message(message_data) ctxt = unpack_context(self.conf, message_data) """解析消息""" method = message_data.get('method') args = message_data.get('args', {}) version = message_data.get('version') namespace = message_data.get('namespace') 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 """处理消息""" self.pool.spawn_n(self._process_data, ctxt, version, method, namespace, args)def _process_data(self, ctxt, version, method, namespace, args): ctxt.update_store() try: rval = self.proxy.dispatch(ctxt, version, method, namespace, **args) ... 可以看到消息的处理主要是通过新开的协程（使用eventlet）来执行self._process_data函数，而self._process_data函数中主要调用了self.proxy.dispatch函数，那么这个dispatch函数的内容究竟是什么呢？
1、首先找到self.proxy
根据对self.proxy的层层追踪，其为上面ProxyCallback构造时传入，即start函数中的 rpc_dispatcher = self.manager.create_rpc_dispatcher(self.backdoor_port)，这里的self.manager在创建该Service的时候就已经设定为ConductorManager。ConductorManager的create_rpc_dispatcher函数代码如下：
def create_rpc_dispatcher(self, *args, **kwargs): #self.compute_task_mgr = ComputeTaskManager() kwargs['additional_apis'] = [self.compute_task_mgr] return super(ConductorManager, self).create_rpc_dispatcher(*args, **kwargs)
根据类的继承关系：
nova/manager.py
Manager类：def create_rpc_dispatcher(self, backdoor_port=None, additional_apis=None): apis = [] if additional_apis: apis.extend(additional_apis) base_rpc = baserpc.BaseRPCAPI(self.service_name, backdoor_port) apis.extend([self, base_rpc]) serializer = objects_base.NovaObjectSerializer() return rpc_dispatcher.RpcDispatcher(apis, serializer)
nova/openstack/common/rpc/dispatcher.py
RpcDispatcher类def __init__(self, callbacks, serializer=None): #callbacks为一个list，包含ComputeTaskManager,ConductorManager,BaseRPCAPI self.callbacks = callbacks if serializer is None: serializer = rpc_serializer.NoOpSerializer() self.serializer = serializer super(RpcDispatcher, self).__init__()
所以最后的self.proxy为RpcDispatcher对象。要注意的是RpcDispatcher对象的callbacks属性包含了一系列的Manager类，它们将被用来处理消息中指定的method。参考下图：
2、分析self.proxy.dispatch函数
即RpcDispatcher类的dispatch函数，代码如下：
def dispatch(self, ctxt, version, method, namespace, **kwargs): if not version: version = '1.0' had_compatible = False for proxyobj in self.callbacks: try: cb_namespace = proxyobj.RPC_API_NAMESPACE except AttributeError: cb_namespace = None if namespace != cb_namespace: continue ... if not hasattr(proxyobj, method): continue if is_compatible: kwargs = self._deserialize_args(ctxt, kwargs) result = getattr(proxyobj, method)(ctxt, **kwargs) return self.serializer.serialize_entity(ctxt, result) ...
该函数会根据namespace匹配Manager，然后分析Manager类中的函数是否和消息中的函数匹配，如果匹配即调用Manager类中对应的函数进行处理。经过观察，其实对外提供rpc服务的组件的rpcapi.py和manager.py文件中的类是对应的，它们通过namespace进行匹配查找，在命名上具有相同的前缀，所以如果要跟踪rpcapi.py中函数的后续实现只需查看该组件的manager.py文件里对应的Manager类的对应函数就可以了。
参考文章：