openstack oslo.messaging库
2017年04月13日 22:13:25 li_101357 阅读数:1383根据OpenStack官网介绍,oslo.messaging库就是把rabbitmq的Python库做了封装,在openstack中调用RPC通信就要调用oslo.messaging库,为RPC和事件通知提供一套统一的接口。
下面介绍oslo.messaging在RPC通信过程中两个重要部分,RPC Client,Server。
Server是RPC服务器提供多个端点,每个包含一组远程调用客户端的方法,创建一个PRC服务器,提供一个传输队列,目标和端点列表。
Client是一个类调用远程服务器上的方法,RPCClient类是负责发送方法调用通过消息传输到远程服务器。
一、oslo.messaging对如下的对象进行统一
1、Transport
Transport(传输层)主要实现RPC底层的通信(比如socket)以及事件循环、多线程等其他功能。可以通过URL来获得指向不同Transport实现的句柄。URL格式:
Transport://user:pass@host1:port[,hostN:portN]/virtual_host
从cfg对象读取transport_url、rpc_backend、control_exchange信息构造Transport对象。用户可以使用oslo.messaging.get_transport函数来获得transport对象实例的句柄。
2、Target
Target封装指定某一个消息最终目的地的所有信息。
Target对象属性
参数 = 默认值 |
说 明 |
exchange=None |
(字符串类型)交换器,topic所属的范围,默认使用配置文件中的control_exchange选项 |
topic=None |
(字符串类型)一个topic表示服务器暴露的一组接口(可被远程调用的方法)。 允许多个服务器暴露同一组接口,消息会轮询发给多个服务器中的某一个 |
namespace=None |
(字符串类型)表示服务器暴露的某个特定的接口(多个可被远程调用的方法) |
version=None |
服务器所暴露的接口支持M.N类型的版本号 |
server=None |
(字符串类型)客户端可以指定 此参数要求消息的目的地是某个特定的服务器, 而不是 topic 一组服务器的一台。 |
fanout=None |
(布尔类型)当设置为真时,消息会被广播发送给 同属某个topic的所有服务器,而不是 |
注意:不同场景下构造Target对象需要不同的参数:
(1)、创建RPC服务器时,需要topic和server参数,exchange参数可选;指定一个endpoint的target时,namespace和version是可选的
(2)、客户端发送消息时,需要topic参数,其他可选
3、Server
一个RPC服务器可以暴露多个endpoint,每个endpoint包含一组方法,这组方法是可以被客户端通过transport对象远程调用的。 创建Server对象时,需要指定Transport、Target和一组endpoint。
4、RPC Client
远程调用RPC Server上的方法。远程调用的时候,需要提供一个字典对象指明调用的上下文,调用方法的名字和传递给调用方法的参数。
调用方式:(1)、通过cast方式 异步远程调用,请求发送之后客户端直接返回。
(2)、call方式 同步远程调用,等待服务器响应返回结果。
程序案例:
app.conf:
- [DEFAULT]
- username=app
- [rabbit]
- host = 192.168.1.7
- port = 5672
- server.py: 服务器端程序
- from oslo_config import cfg
- import oslo_messaging
- from oslo_log import log as logging
- import time
- CONF = cfg.CONF
- LOG = logging.getLogger(__name__)
- logging.register_options(CONF)
- logging.setup(CONF, "myservice")
- CONF(default_config_files=['app.conf'])
- class ServerControlEndpoint(object):
- target = oslo_messaging.Target(namespace='control',
- 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):
- print "test"
- print arg
- return arg
- transport = oslo_messaging.get_transport(cfg.CONF)
- target = oslo_messaging.Target(topic='test123', server='server1')
- endpoints = [
- ServerControlEndpoint(None),
- TestEndpoint(),
- ]
- server=oslo_messaging.get_rpc_server(transport,target, endpoints,executor='blocking')
- try:
- server.start()
- while True:
- time.sleep(1)
- except KeyboardInterrupt:
- print("Stopping server")
- server.stop()
- server.wait()
(1)、服务器定义两个不同的endpoint:ServerControlEndpoint和TestEndpoint。这两个endpoint中的方法stop和test都可以被客户端远程调用。
(2)、创建rpc server之前,需要先创建transport和target对象,使用get_transport函数获得transport对象的句柄,get_transport函数的参数如下:
参数=默认值 |
说明 |
conf |
oslo.config配置项对象 |
url=None |
(字符串或oslo.messaging.Transport类型) transportURL。为空则采用conf配置中的transporturl项所指定的值 |
namespace=None |
(字符串类型)用来标识服务器所暴露的某个特定接口 |
allowed_remote_exmods=None |
|
aliases=None |
(字典类型)transport别名和transport名称之间的对应关系 |
(3)、Target对象是用来建立RPCserver的,需要指明topic和server参数。用户定义的endpoint对象可以包含一个target属性,来指明这个endpoint所支持的特定的namespace和version.
(4)、get_rpc_server()函数创建server对象,然后调用server对象的start方法开始接受远程调用,get_rpc_server()函数参数如下示:
参数=默认值 |
说明 |
transport |
(Transport类型)transport对象 |
target |
(target类型)指明监听的exchange、topic和server |
endpoints |
(列表类型)endpoint对象实例的列表 |
executor=’blocking’ |
(字符串类型)指明消息的接收和处理的方式 (1) blocking:用户调用start函数后,在start函数中开始循环处理请求:用户线程阻塞,处理下一个请求。直到用户调用stop函数之后处理循环才会退出。 消息的分发和处理循环都是在start函数的线程中完成。 (2) eventlet:会有协程GreenThread来处理消息的接收,然后会有不同的协程来处理不同消息的分发处理。调用start函数的用户线程不会阻塞 |
Serializer=None |
用来序列化/反序列化消息 |
- client.py:
- import oslo_messaging as messaging
- from oslo_context import context
- from oslo_config import cfg
- from oslo_log import log as logging
- CONF = cfg.CONF
- LOG = logging.getLogger(__name__)
- logging.register_options(CONF)
- logging.setup(CONF, "myservice")
- CONF(default_config_files=['app.conf'])
- ctxt = {}
- arg = {'a':'b'}
- transport = messaging.get_transport(cfg.CONF)
- target = messaging.Target(topic='test123')
- client = messaging.RPCClient(transport, target)
- client.call(ctxt, 'test', arg=arg)
- cctxt = client.prepare(namespace=’control’,version=’2.0’)
- cctxt.cast({}, ‘stop’)
(1)、远程调用时,需要传入字典对象指明调用上下文、调用方法名字和传给调用方法的参数。
(2)、Target对象的属性在RPCClient对象构造之后,还可以通过prepare()方法修改,可以修改的属性包括exchange,topic,namespace,version,server,fanout和retry。修改后的target属性只在这个prepare()方法返回的对象中有效。
程序运行结果:call同步方式调用
客户端同步调用返回结果。