oslo.messaging
oslo.messaging库为OpenStack各个项目使用RPC和事件通知（Event Notification)提供了一套统一的接口.代码库位于https://github.com/openstack/oslo.messaging,项目主页为https://launchapd.net/oslo.messaging,参考文档在http://docs.openstack.org/developer/oslo.messaging.
为了支持不同的RPC后端实现,oslo.messaging对如下的对象进行了统一:
- Transport
Transport(传输层)主要实现RPC底层的通信(比如socket)以及事件循环,多线程等其他功能.可以通过URL来获得不同transport的句柄.URL的格式为:
transport://user:password@host:port[,hostN:portN]/virtual_host
目前支持的Transport有rabbit,qpid与zmq,分别对应不同的后端消息总线.用户可以使用oslo.messaging.get_transport函数来获得transport对象实例的句柄.
- Target
Target封装了指定某一个消息最终目的地的所有信息,下表所示为其所具有的属性:
| 参数=默认值 | 说 明 |
| exchange = None | (字符串类型)topic所属的范围,不指定的话默认使用配置文件中的control_exchange选项 |
| topic = None | (字符串类型)一个topic可以用来标识服务器所暴露的一组接口(一个接口包含多个可被远程调用的方法).允许多个服务器暴露同一组接口,消息会以轮循的方式发送给多个服务器中的某一个 |
| namespace = None | (字符串类型)用来标识服务器所暴露的某个特定接口(多个可被远程调用的方法) |
| version = None | (字符串类型)服务器所暴露的接口支持M.N类型的版本号.次版本号(N)的增加表示新的接口向前兼容,主版本号(M)的增加表示新接口和旧接口不兼容.RPC服务器可以实现多个不同的主版本号接口. |
| server = None | (字符串类型)客户端可以指定此参数来要求消息的目的地是某个特定的服务器,而不是一组同属某个topic的服务器中的任意一台. |
| fanout = None | (布尔型)当设置为真时,消息会被发送到同属某个topic的所有服务器上,而不是其中的一台. |
在不同的应用场景下,构造Target对象需要不同的参数:创建一个RPC服务器时,需要topic和server参数,exchange参数可选;指定一个endpoint时,namespace和version是可选的;客户端发送消息时,需要topic参数,其他可选.
- Server
一个RPC服务器可以暴露多个endpoint,每个endpoint包含一组方法,这组方法是可以被客户端通过某种Transport对象远程调用的.创建Server对象时,需要指定Transport,Target和一组endpoint.
- RPC Client
通过RPC Client,可以远程调用RPC Sever上的方法.远程调用时,需要提供一个字典对象来指明调用的上下文,调用方法的名字和传递给调用方法的参数(用字典表示).
有cast和call两种远程调用方式.通过cast方式远程调用,请求发送后就直接返回了;通过call方式调用,需要等响应从服务器返回.
- Notifier
Notifier用来通过某种transport发送通知消息.通知消息遵循如下的格式:
import six import uuid from oslo_utils import timeutils {'message_id': six.text_type(uuid.uuid4()), #消息id号 'publisher_id': 'compute.hos1', #发送者id 'timestamp': timeutils.utcnow(), #时间戳 'priority': 'WARN', #通知优先级 'event_type': 'compute.create_instance', #通知类型 'payload': {'instance_id': 12, ...}} #通知内容
可以在不同的优先级别上发送通知,这些优先级包括sample,critical,error,warn,info,debug,audit等.
- Notification Listener
import oslo_messaging class ErrorEndpoint: def error(self, ctxt, publisher_id, event_type, payload, metadata): do_something(payload) return oslo_messaging.NotificationResult.HANDLED
endpoint中的方法如果返回messaging.NotificationResult.HANDLED或者None,表示这个通知消息已经确认被处理;如果返回messaging.NotificationResult.REQUEUE,表示这个通知消息要重新进入消息队列.
下面是一个利用oslo_messaging来实现远程过程调用的示例.
from oslo_config import cfg import oslo_messaging as messaging class ServerControlEndpoint(object): target = messaging.Target(namespace='controle', version='2.0') def __init__(self, server): self.server = server def stop(self, ctx): if self.server: self.server.stop() class TestEndpoint(object): def test(self, ctx, arg): return arg transport = messaging.get_transport(cfg.CONF) target = messaging.Target(topic='test', server='server1') endpoints = [ ServerControlEndpoint(None), TestEndpoint(), ] server = messaging.get_rpc_server(transport, target, endpoints, executor='blocking') server.start() server.wait()
这个例子里,定义了两个不同的endpoint:ServerControlEndpoint与TestEndpoint.这两个endpoint中的方法stop和test都可以被客户端远程调用.
创建rpc server对象之前,需要先创建transport和target对象,这里使用get_transport()函数来获得transport对象的句柄,get_transport()的参数如下表所示:
| 参数=默认值 | 说 明 |
| conf | (oslo.config.cfg.ConfigOpts类型)oslo.config配置项对象 |
| url = None | (字符串或者oslo.messaging.Transport类型)transport URL.如果为空,采用conf配置中的transport_url项所指定的值 |
| namespace = None | (字符串类型)用来标识服务器所暴露的某个特定接口(多个可被远程调用的方法) |
| allowed_remote_exmods = None | (列表类型)Python模块的列表.客户端可用列表里的模块来deserialize异常 |
| aliases = None | (字典类型)transport别名和transport名称之间的对应关系 |
conf对象里,除了包含transport_url项外,还可以包含control_exchange项.control_exchange用来指明topic所属的默认范围,默认为"openstack".可以使用oslo.messaging.set_transport_defaults()函数来修改默认值.
此处构建的Target对象是用来建立RPC Server的,所以需指明topic和server参数.用户定义的endpoint对象也可以包含一个target属性,用来指明这个endpoint所支持的特定的namespace和version.
这里使用get_rpc_server()函数创建server对象,然后调用server对象的start方法开始接收远程调用.get_rpc_server()函数的参数如下表所求:
| 参数=默认值 | 说 明 |
| transport | (Transpor类型)transport对象 |
| target | (Target类型)target对象,用来指明监听的exchange,topic和server |
| endpoints | (列表类型)包含了endpoints对象实例的列表 |
| executor='blocking' | (字符串类型)用来指明消息接收和发收的方式:目前支持两种方式: blocking:在这种方式中,用户调用start函数后,在start函数中开始请求处理循环:用户线程阻塞,处理下一个请求.直到用户调用了stop函数后,这个处理循环才会退出.消息的接收和分发处理都在调用start函数的线程中完成. eventlet:在这种方式中,会有一个协程GreenThread来处理消息的接收,然后有其他不同的GreenThread来处理不同消息的分发处理.调用start的用户线程不会被阻塞 |
| serializer = None | (Serializer类型)用来序列化/反序列化消息 |
#client.py 客户端 from oslo_config import cfg import oslo_messaging as messaging transport = messaging.get_transport(cfg.CONF) target = messaging.Target(topic='test') client = messaging.RPCClient(transport, target) ret = client.call(ctxt={}, method='test', arg='myarg') cctx = client.prepare(namespace='control', version='2.0') cctx.cast({}, 'stop')
这里target对象构造时,必须要有的参数只有topic,创建RPCClient对象时,可以接受的参数如下表所示:
| 参数=默认值 | 说 明 |
| transport | (Transport类型)transport对象 |
| target | (Taget类型)该client对象的默认target对象 |
| timeout = None | (整数或者浮点数类型)客户端调用call方法时超时时间(秒) |
| version_cap = None | (字符串类型)最大所支持的版本号.当版本号超过时,会扔出RPCVersionCapError异常 |
| serializer = None | （Serializer类型)用来序列化/反序列化消息 |
| retry = None |
(整数类型)连接重试次数:
None或者-1:一直重试
0:不重试
>0:重试次数
|
远程调用时,需要传入调用上下文,调用方法的名字和传给调用方法的参数.
Target对象的属性在RPCClient对象构造以后,还可以通过prepare()方法修改.可以修改的属性包括exchange,topic,namespace,version,server,fanout,timeout,version_cap和retry.
修改后的target属性只在这个prepare()方法返回的对象中有效.
下面我们再来看一个利用oslo_messaing实现通知消息处理的例子:
#notification_listener.py 消息通知处理 from oslo_config import cfg import oslo_messaging as messaging class NotificationEndPoint(object): def warn(self, ctxt, publisher_id, event_type, payload, metadata): do_something(payload) class ErrorEndpoint(object): def error(self, ctxt, publisher_id, event_type, payload, metadata): do_something(payload) transport = messaging.get_transport(cfg.CONF) targets = [ messaging.Target(topic='notifications'), messaging.Target(topic='notifications_bis') ] endpoints = [ NotificationEndPoint(), ErrorEndpoint(), ] listener = messaging.get_notification_listener(transport, targets, endpoints) listener.start() listener.wait()
通知消息处理的endpoint对象和远程过程调用的endpoint对象不同,对象定义的方法要和通知消息的优先级一一对应.我们可以为每个endpoint指定所对应的target对象.
最后调用get_notificaton_listener()函数构造notification listener对象,get_notification_listener()函数的参数如下表所示:
| 参数=默认值 | 说 明 |
| transport | (Transport类型)transport对象 |
| target | (列表类型)target对象的列表,用来指明endpoints列表中的每一个endpoint所侦听处理的exchange和topic |
| endpoints | (列表类型)包含了endpoints对象实例的列表 |
| executor='blocking' | (字符串类型)用来指明消息接收和发收的方式:目前支持两种方式: blocking:在这种方式中,用户调用start函数后,在start函数中开始请求处理循环:用户线程阻塞,处理下一个请求.直到用户调用了stop函数后,这个处理循环才会退出.消息的接收和分发处理都在调用start函数的线程中完成. eventlet:在这种方式中,会有一个协程GreenThread来处理消息的接收,然后有其他不同的GreenThread来处理不同消息的分发处理.调用start的用户线程不会被阻塞 |
| serializer=None | （Serializer类型）用来序列化/反序列化消息 |
| allow_requeue=False | (布尔类型)如果为真,表示支持NotificationResult.REQUEUE |
相对应的发送消息通知的代码如下:
#notifier_send.py from oslo_config import cfg import oslo_messaging as messaging transport = messaging.get_transport(cfg.CONF) notifier = messaging.Notifier(transport, driver='messaging', topic='notifications') notifier2 = notifier.prepare(publisher_id='compute') notifier2.error(ctxt={}, event_type='my_type', payload={'content': 'error occurred'})
发送通知消息时,首先要构造Notifier对象,此时可能需要指定的参数如下表所示:
| 参数=默认值 | 说 明 |
| transport | （Transport类型)transport对象 |
| target | (列表类型)target对象的列表,用来指明endpoints列表中的每一个endpoint所侦听处理的exchange和topic |
| publish_id = None | (字符串类型)发送者id |
| driver = None | (字符串类型)后台驱动.一般采用"messaging".如果没有指定,会使用配置文件中的notificaton_driver的值 |
| topic = None | (字符串类型)发送消息的topic.如果没有指定,会使用配置文件中的notification_topics的值 |
| serializer = None | （Serializer类型）用来序列化/反序列化消息 |
初始化Notifier对象的操作比较复杂,所以可以用prepare()方法修改已创建的Notifier对象,prepare()方法返回的是新的Notifier对象的实例.它的参数如下表所示:
| 参数 = 默认值 | 说 明 |
| publish_id = None | (字符串类型)发送者id |
| retry = None |
(整数类型)连接重试次数:
None或者-1:一直重试
0:不重试
>0:重试次数
|
最后可以调用Notifier对象的不同方法(error, critical, warn, 等等)发送不同优先级的消息通知.