說完了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發布消息。

總結

以上是生活随笔為你收集整理的notification源码分析_Ceilometer之notification agent代码分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。