Openstack基础组件之oslo.messaging

1. Transport

传输消息的工具，比如RabbitMQ等一些常见的消息队列，通常获取的途径是通过用户直接配置transport_url调用接口来获取：

import oslo_messaging
transport = oslo_messaging.get_transport(cfg, url=None, **kwargs)

2. Target

用来标识消息的目的地，会对消息进行封装，决定消息将发送到何处或者其他Server将会监听什么类型的消息，它的对象类的定义如下：
class oslo_messaging.Target(exchange=None, topic=None, namespace=None, version=None, server=None, fanout=None, legacy_namespaces=None)

而对于不同封装在Target的子类消息对象可能需要的初始化参数不一样：

Target类型	Required Params	Optional Params
RPC Server Target	topic、server	exchange
RPC Endpoint		namespace、version
RPC Client	topic	all other attributes
Notification Server Target	topic	exchange, all other attributes ignored
Notifier Target	topic	exchange, all other attributes ignored

3. RPC

3.1 RPC Server

一个RPC Server的创建需要依赖至少三个部分：Transport、Target和Endpoints(这里的Endpoints可以理解成一个方法的集合类，这些方法会用来被RPC Client远程调用)，下面是一个RPC服务的类伪代码实现，包括RPC Server的创建和启动等：

class test1Endpoint(object):
  def test_func01(self, ctxt, *args):
    pass

class test2Endpoint(object):
  def test_func02(self, ctxt, *args):
    pass

class RPCService(os_service.Service):
  def __init__(self):
    super(RPCService, self).__init__()
    endpoints = [test1Endpoint(), test2Endpoint()]
    transport = oslo_messaging.get_transport(conf, url=None)
    target = osl_messaging.Target(server='host1', topic=topic)
    self.rpc_server = oslo_messaging.get_rpc_server(transport, target, endpoints, executor='eventlet')

  def start(self):
    super(RPCService, self).start()
    self.rpc_server.start()
    self.tg.add_timer(604800, lambda: None)

  def stop(self):
    self.rpc_server.stop()
    super(RPCService, self).stop()

3.2 RPC Client

调用远程RPC Server方法的类，如下是实现一个RPC Client的伪代码：

class RPCClient(object):
  def __init__(self):
    transport = oslo_messaging.get_transport(conf, url=None)
    target = osl_messaging.Target(topic=topic)
    self.rpc_client = oslo_messing.RPCClient(transport, target)

  def func01(self, ctxt, *args):
    return self.rpc_client.call(ctxt, 'test_func01', args=args)
    # self.rpc_client.cast(ctxt, 'test_func01', args=args)  # call等待直至返回一个结果，cast不等待立即返回

  def func02(self, ctxt, *args):
    cctxt = self.rpc_client.prepare(server='host01', version='4.0')
    return cctxt.call(ctxt, 'test_func02', args=args)

4. Notification

4.1 Notifier

功能：发送通知消息，消息的格式如下：

{'message_id': six.text_type(uuid.uuid4()),
 'publisher_id': 'compute.host1',
 'timestamp': timeutils.utcnow(),
 'priority': 'WARN',
 'event_type': 'compute.create_instance',
 'payload': {'instance_id': 12, ... }}

构建Notifier的类：
class oslo_messaging.Notifier(transport, publisher_id=None, driver=None, serializer=None, retry=None, topics=None)

具体实现流程：创建或者获取Notifier --> Notifier发送不同级别的消息，下面是一个简单实现的伪代码：

transport = oslo_messaging.get_transport(conf)
notifier = oslo_messaging.Notifier(transport, publisher_id='host1', driver='messagingv2', topics=['notifications'])
# notifier = notifier.prepare(publisher_id='host0')  # 可以通过这种方式对Notifier的某些属性进行覆盖，跟RPC Client的prepare功能相同
notifier.info(ctxt, 'compute.instance.create.start', payload)  # 第二个参数是event_type，payload则是跟event_type对应任务相关的一些有用数据信息

4.2 Notification Driver

常见的Notification Driver有log、messaging、messagingv2等，比如常用消息队列来发送通知消息，而它具体driver出自oslo_messaging.notify.messaging

4.3 Notification Listener

功能：接收和处理通知消息，这一部分一般至少需要两个部分的实现：Notification Listener创建和Endpoints(会实现一些方法，跟Notifier端发送消息的级别对应，实现获取通知消息后如何处理及留存)，如下是一个简单的伪代码实现：

class Endpoint(object):
  def __init__(self):
    pass

  def info(self, ctxt, publisher_id, event_type, payload, metadata):
    pass

class NotificationService(os_service.Service)
  def __init__(self):
    super(NotificationService, self).__init__()
    self.listeners = []

  def start(self):
    super(NotificationService, self).start()
    transport = oslo_messaging.get_transport(conf)
    target = oslo_messaging.Target(topic='notifications')
    endpoints = [Endpoint()]
    listener = oslo_messaging.get_notification_listener(transport, target, endpoints, executor='eventlet', pool='pool0')
    listener.start()
    self.listeners.append(listener)
    self.tg.add_timer(604800, lambda: None)

  def stop(self):
    if self.listeners:
      map(lambda: x: x.stop(), self.listeners)
    super(NotificationService, self).stop()

5. 参考资料

Openstack oslo_messaging官方文档