import sysfrom nova.cmd.conductor import mainif __name__ == "__main__":    sys.exit(main())

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()

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

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)

def launch(service, workers=None):    if workers:        launcher = ProcessLauncher()        launcher.launch_service(service, workers=workers)    else:        launcher = ServiceLauncher()        launcher.launch_service(service)    return launcher

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)

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

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)    ...

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)

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)

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__()

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)    ...