openstack nova 基础知识——从源码看一个服务是如何启动的

nova中服务分为两种：Service和WSGIService，后者我还不清楚是用来做什么工作的，但是前者看了一下bin中的代码，知道了它是scheduler, compute, network等这些组件所使用的服务。

相对这两种服务，就有两种加载的方法：ServiceLauncher和ProcessLauncher，看代码这两者的主要区别是前者放到了绿色线程中，后者则是用os.fork()出来的进程。

看一下整体的类图：



下面就以Scheduler服务的启动为例，来看一下启动的过程：

server = service.Service.create(binary='nova-scheduler')
service.serve(server)
service.wait()

1.  第一行是调用Service中的类方法create()创建了一个Service对象，这里仅仅是创建了一个对象，所做的工作是初始化类属性，比如这里只传递了一个binary参数进去，但是一个Service不可能只需要这一个参数，那么其它的参数就是要从配置文件中读取或是使用默认的值。来看一下create()方法：

@classmethod
def create(cls, host=None, binary=None, topic=None, manager=None,
report_interval=None, periodic_interval=None,
periodic_fuzzy_delay=None):
"""Instantiates class and passes back application object.

:param host: defaults to FLAGS.host
:param binary: defaults to basename of executable
:param topic: defaults to bin_name - 'nova-' part
:param manager: defaults to FLAGS.<topic>_manager
:param report_interval: defaults to FLAGS.report_interval
:param periodic_interval: defaults to FLAGS.periodic_interval
:param periodic_fuzzy_delay: defaults to FLAGS.periodic_fuzzy_delay

"""
if not host:
host = FLAGS.host #host='ubuntu'
if not binary: #binary='nova-scheduler'
binary = os.path.basename(inspect.stack()[-1][1])
if not topic:
topic = binary.rpartition('nova-')[2] #topic='scheduler'
if not manager: #manager='nova.scheduler.manager.SchedulerManager'
manager = FLAGS.get('%s_manager' % topic, None)
if report_interval is None:
report_interval = FLAGS.report_interval
if periodic_interval is None:
periodic_interval = FLAGS.periodic_interval
if periodic_fuzzy_delay is None:
periodic_fuzzy_delay = FLAGS.periodic_fuzzy_delay
service_obj = cls(host, binary, topic, manager,
report_interval=report_interval,
periodic_interval=periodic_interval,
periodic_fuzzy_delay=periodic_fuzzy_delay)

return service_obj

从配置文件读取出来其它参数之后，调用cls()来真正的实例化一个类，注意这里create()是一个类方法，传递进来的第一个参数cls就是Service类本身。

2.  创建好Service对象之后，就是加载服务了，那就是上面第二行所做的事，Scheduler服务是用ServiceLauncher来加载的：

_launcher = ServiceLauncher()
_launcher.launch_server(server)

def launch_server(self, server):
"""Load and start the given server.

:param server: The server you would like to start.
:returns: None

"""
gt = eventlet.spawn(self.run_server, server)
self._services.append(gt)

@staticmethod
def run_server(server):
"""Start and wait for a server to finish.

:param service: Server to run and wait for.
:returns: None

"""
server.start()#尼妈啊，终于找到你了，原来是在这调用的start()，太隐蔽了啊！！！
server.wait()

从launcher_server()中可以看到加载一个服务是放到了绿色线程中，执行这个run_server()方法。run_server()方法执行server.start()和server.wait()，完成了服务的启动，然后等待，一开始我看到Service类中有个start()方法，知道一个服务的启动实质性的内容都在这个方法里，但是就是找不到在哪里调用了这个方法，原来是在这里。那么接下来就进到这个start()中了。

3.  start()主要做的工作是从数据库中取得服务的相关信息，还有创建了三个不同类型的Consumer，用来接收消息，看一下代码：

def start(self):
vcs_string = version.version_string_with_vcs()
LOG.audit(_('Starting %(topic)s node (version %(vcs_string)s)'),
{'topic': self.topic, 'vcs_string': vcs_string})
utils.cleanup_file_locks()
self.manager.init_host()
self.model_disconnected = False
ctxt = context.get_admin_context()#返回一个RequestContext对象，并且属性is_admin=True
try:
# 得到在‘host’主机中的‘binary’的‘service’在数据库中保存的状态
service_ref = db.service_get_by_args(ctxt,
self.host,
self.binary)
self.service_id = service_ref['id'] #得到服务的id，因为models.Service有迭代器的性质
except exception.NotFound:
self._create_service_ref(ctxt)

if 'nova-compute' == self.binary:
self.manager.update_available_resource(ctxt)

# 从连接池中取得一个连接
self.conn = rpc.create_connection(new=True)
LOG.debug(_("Creating Consumer connection for Service %s") %
self.topic)

rpc_dispatcher = self.manager.create_rpc_dispatcher()

# Share this same connection for these Consumers
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)

# Consume from all consumers in a thread
self.conn.consume_in_thread()#把consumer放到绿色线程中，并且启动监听消息

if self.report_interval:
pulse = utils.LoopingCall(self.report_state)
pulse.start(interval=self.report_interval,
initial_delay=self.report_interval)
self.timers.append(pulse)

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

periodic = utils.LoopingCall(self.periodic_tasks)
periodic.start(interval=self.periodic_interval,
initial_delay=initial_delay)
self.timers.append(periodic)

主要来看一下创建Consumer，创建Consumer其实就是创建Queue，并且声明topic，告诉Exchange我要接收那些消息，这里定义了三个Consumer，两个Topic类型的Consumer和一个Fanout类型的Consumer，然后通过consumer_in_thread()把consumer的监听程序(drain_event())放到了绿色线程中。

这样一个服务就启动了，然后调用server.wait()来等待服务结束。。