oslo_messaging源于Openstack的一个经典的模块，用以实现服务间的RPC通信。Client端将数据放入rabbitmq中，server端从消息队列中获取传送数据。
oslo.messaging库就是把rabbitmq的python库做了封装，考虑到了编程友好、性能、可靠性、异常的捕获等诸多因素。让各个项目的开发者聚焦于业务代码的编写，而不用考虑消息如何发送和接收。
一张比较经典的图见下：
Target：作为消息发送者，需要在target中指定消息要发送到的topic，exchange, binding-key, consumer等信息。
Transport(传输层)主要实现RPC底层的通信(比如socket)以及事件循环,多线程等其他功能.可以通过URL来获得不同transport的句柄.URL的格式为:
transport://user:password@host:port[,hostN:portN]/virtual_host
目前支持的Transport有rabbit,qpid与zmq,分别对应不同的后端消息总线.用户可以使用oslo.messaging.get_transport函数来获得transport对象实例的句柄.
Notifier：消息的发送端，可以在不同的优先级别上发送通知,这些优先级包括sample,critical,error,warn,info,debug,audit等
Notification Listener和Server类似,一个Notification Listener对象可以暴露多个endpoint,每个endpoint包含一组方法.但是与Server对象中的endpoint不同的是,这里的endpoint中的方法对应通知消息的不同优先级。在发送消息时，指定方法info，warn等，在notifer listener监听消息队列，使用dispatcher对象根据消息的publish_id, event_type将消息路由到不同的endpoint方法上。
举个例子，在notifier listener端程序见下：
from oslo_config import cfg
import oslo_messaging class NotificationEndpoint(object):
# filter_rule = oslo_messaging.NotificationFilter(
# publish_id='^compute.*')
def warn(self, ctxt, publish_id, event_type, payload, metadata):
print "caesar==> %s" % payload class ErrorEndpoint(object):
# filter_rule = oslo_messaging.NotificationFilter(
# event_type='^instance\..*\.start',
# context={'ctxt_key':'regexp'}) def error(self, ctxt, publish_id, event_type, payload, metadata):
print "caesar==> %s" % payload transport = oslo_messaging.get_notification_transport(cfg.CONF)
endpoints = [
NotificationEndpoint(),
ErrorEndpoint()
]
targets = [
oslo_messaging.Target(topic='notification'),
oslo_messaging.Target(topic='notification_bis')
] server = oslo_messaging.get_notification_listener(transport, targets,
endpoints)
server.start()
server.wait()
程序中，两个endpoint中分别有error和warn方法，当开启服务时，会创建四个topic消息 队列，见下：
在客户端，通过notifier中topic和方法，比如topic=notification 方法为error，即可以向notification.error队列中传入数据。
from oslo_config import cfg
import oslo_messaging as messaging transport = messaging.get_transport(cfg.CONF)
notifier = messaging.Notifier(transport, driver='messaging', topics=['notification'])
project_id = 'b23a5e41d1af4c20974bf58b4dff8e5a'
user_id = 'ceb61464a3d341ebabdf97d1d4b97099'
notifier.error(ctxt={},
event_type='my_type',
payload={
'tenant_id': project_id,
'user_id': user_id,
'instance_id': '',
'instance_type_id': 1,
'instance_type': 'm1.flavor',
'state': 'active' })
执行notifier程序，查询消息队列为空，即已经被notification listnener消费，消息无阻塞。：
在notification listnener 路由到ErrorEndpoint的error方法，打印结果见下：
