说完了polling agent，咱们接着说notification agent，打开setup.cfg文件，找到入口 ceilometer-agent-notification = ceilometer.cmd.agent_notification:main，打开文件，发现启动了NotificationService服务，然后看start方法，核心代码如下
[mw_shl_code=applescript,true]self.listeners, self.pipeline_listeners = [], []
self._configure_main_queue_listeners(transporter, event_transporter)[/mw_shl_code]
意思是配置主消息队列监听器(主消息队列是指exchange为nova,cinder,neutron等的消息队列)，来到该方法下，
[mw_shl_code=applescript,true]notification_manager = self._get_notifications_manager(transporter)
for ext in notification_manager:
handler = ext.obj
for new_tar in handler.get_targets(cfg.CONF):
if new_tar not in targets:
targets.append(new_tar)
endpoints.append(handler)
urls = cfg.CONF.notification.messaging_urls or [None]
for url in urls:
transport = messaging.get_transport(url)
listener = messaging.get_notification_listener(
transport, targets, endpoints)
listener.start()
self.listeners.append(listener)
[/mw_shl_code]
首先调用命名空间是ceilometer.notification的plugins(还在setup.cfg中)，然后获取相关的endpoins和targets，最后启动监听器，最重要的是启动了监听器
我们不断找监听器的代码，最后发现是在oslo_messaging中的server.py文件中，看他的start方法
[mw_shl_code=applescript,true]if self._executor is not None:
return
try:
listener = self.dispatcher._listen(self.transport)
except driver_base.TransportDriverError as ex:
raise ServerListenError(self.target, ex)
self._executor = self._executor_cls(self.conf, listener,
self.dispatcher)
self._executor.start()[/mw_shl_code]
这里还是用到了plugin，命名空间是oslo.messaging.executors，调用执行器执行代码，我们debug获得plugin是哪个，结果是eventlet，找到他的代码，
文件在/usr/lib/python2.7/site-packages/oslo_messaging/_executors/impl_eventlet.py中，看他的start方法，核心如下
[mw_shl_code=applescript,true]@excutils.forever_retry_uncaught_exceptions
def _executor_thread():
try:
while self._running:
incoming = self.listener.poll()
if incoming is not None:
self._dispatch(incoming)
except greenlet.GreenletExit:
return[/mw_shl_code]
意思就是不断调用listener的poll方法获得incoming，(意思就是不断获取消息队列上的消息)，如果有，就执行self._dispatch(incoming)
我们找到listener，
[mw_shl_code=applescript,true]listener = AMQPListener(self, conn)
for target, priority in targets_and_priorities:
conn.declare_topic_consumer(
exchange_name=self._get_exchange(target),
topic='%s.%s' % (target.topic, priority),
callback=listener, queue_name=pool)
return listener[/mw_shl_code]
可以看出其实就是设置了许多消费者，监听消息队列上的信息，exchange_name就是nova,cinder之类的，topic是notification,priority类似与info之类
设置这样的消费者就监听了主消息队列上的消息，如果获得了消息(比如某事件触发来了compute.instance.update)，然后执行self._dispatch(incoming)，我们找到该方法,
[mw_shl_code=applescript,true]for screen, callback in self._callbacks_by_priority.get(priority, []):
if screen and not screen.match(ctxt, publisher_id, event_type,
metadata, payload):
continue
localcontext.set_local_context(ctxt)
try:
if executor_callback:
ret = executor_callback(callback, ctxt, publisher_id,
event_type, payload, metadata)
else:
ret = callback(ctxt, publisher_id, event_type, payload,
metadata)
ret = NotificationResult.HANDLED if ret is None else ret
if self.allow_requeue and ret == NotificationResult.REQUEUE:
return ret
finally:
localcontext.clear_local_context()
return NotificationResult.HANDLED[/mw_shl_code]
_callbacks_by_priority是filter_rule和endpoint的方法的dict，首先判断是否匹配(事件类型之类的)，找到某个endpoint的方法，执行它即可。该回调方法的作用一般是publish消息。
以compute的notification中的instance.py的Memory为例子，回调方法执行其父类NotificationBase的info方法，
[mw_shl_code=applescript,true]notification = messaging.convert_to_old_notification_format(
'info', ctxt, publisher_id, event_type, payload, metadata)
self.to_samples_and_publish(context.get_admin_context(), notification)[/mw_shl_code]
将消息发布出去即可，但是在发布过程中，需要调用Memory的get_sample方法返回处理过的消息。重要的是，发布消息的途径还是在pipeline中定义的，如果消息匹配到了相应的meter，就会按照pipeline中相应的sink发布消息。
