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的子类消息对象可能需要的初始化参数不一样：

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官方文档