[OpenStack Icehouse] 消息处理架构Oslo/Messaging

在Icehouse中, rpc消息队列相关处理从openstack.common.rpc慢慢的都转移到oslo.messaging上, 现在仅有几个项目没有转移, 将来也会转, 这个架构更合理, 代码结构清晰, 且弱耦合.

本文章主要针对rpc， notify基本没涉及, 有时间总结总结.

基本概念Server: 为各个Client提供RPC接口,它是消息的最终处理者;Client: RPC接口的调用端, 我们常见的cast和call方法就是在这端调用;Exchange: 理解为一个消息交换机， 把消息分类，告诉何种路由到何种queue;Topic: 是一个RPC消息的唯一标识; servers监听这个topic的消息; client负责发出这个topic的消息;Namespace: servers可以在一个topic上，提供多种方法集合， 这些方法集合通过namespace来分开管理;Method: 这个慨念很简单，， 就是函数, 即远程方法调用中的方法;exchange (defaults to CONF.control_exchange)topicserver (optional) 它会使server的标示, 如host or host@backend 等等fanout (defaults to False)这种模式类似于广播， 符合条件的server都要监听并做处理namespace (optional)API version (optional)Use Cases [OpenStack中使用场景]1.存在多个接口版本的server, 随机选择一个处理远程调用的方法

比如我们有多个接口版本的servers在监听某个exchange上某个topic的方法调用， 一旦方法调用， 就会随机选择一个版本的server来监听。

nova-api 调用’nova’ exchange上 topic为’scheduler’ 的’run_instance’方法， 然后，会有一个‘nova-scheduler’服务捕获请求2.存在多个接口版本的server, 特定server处理远程调用的方法

比如我们有多种接口版本的servers在监听某个exchange上某个topic的方法调用， 不同于之前的是， 它要求选择一个特定版本的server来监听。

nova-scheduler 选择 ‘foobar’ host去运行一个instance，那么就调用’nova’ exchange上 topic为’scheduler’ 的’run_instance’方法， 它同时指定要在“foobar”这个server上run3.存在多个接口版本的server, 每个server都要处理远程调用的方法, 即所谓的fanout

比如我们有多种接口版本的servers在监听某个exchange上某个topic的方法调用， 不同于之前的是， 它要求每个server都要监听并运行这个方法。

nova-compute 周期性的在faout模式调用 ‘update_service_capabilities’ 方法， topic为 ‘scheduler’，exchange为’nova’ ， 这样每个nova-scheduler 服务都要捕获并处理它对象关系图

Server端:消息队列的监听会在service启动的时候开启， 比如cinder-volume启动时，会启动MessageHandlingServer( 下面具体介绍)，来监听消息并把消息dispatch到距离的Manager方法中做消息处理.

Client端: 负责消息发出，方法调用的code是在具体API中，如本例的VolumeAPI， 一般存放在rpcapi.py中.

为了进行详细解释，先画出整体的对象依赖图:

消息处理端 [Server端]MessageHandlingServer

这个就是server端，是消息的最终处理者。

Server端的实现有个两个重要的内部概念：dispatchers 和executors， dispatcher专注在消息的加载并调度到合适的方法。 executor是专注在怎样从transport中获得消息并把它分配给dispatcher的策略，是重开一个线程还是在现有线程上继续处理。实现dispatchers可以为rpc或者notification， executors可以为block或者eventlet

transport：消息中间件，rabbitMQ or Qpid or ZeroMQ dispatcher：rpc或者notification executor：block或者eventlet，默认block， 它描述的是I/O策略，是重开一个线程(EventletExecutor类实现)还是在现有线程上继续处理(BlockingExecutor实现)。

MessageHandlingServer的start()方法开始后 ， 就会一直获取消息队列中信息，并把消息分配给dispatcher， 直到stop()方法被调用。

RPCDispatcher

RPC消息调度器

MessageHandlingServer就依赖它完成从消息到具体处理消息的方法的调度。

它主要干了 两件事

1) 它利用Transport类(Transport又利用AMQPDriverBase的listen()) 获得监听器， 并报告给Executor

def listen(self, target):conn = self._get_connection(pooled=False)listener = AMQPListener(self, conn)conn.declare_topic_consumer(target.topic, listener)conn.declare_topic_consumer(‘%s.%s’ % (target.topic, target.server),listener)conn.declare_fanout_consumer(target.topic, listener)return listener以上代码可以看到listener里做了什么， declare了三个consumer， topic, topic.host, fanout, 正好呼应上面介绍的三种user case。关于consumer， 将会在另一遍博文(介绍amqp的publisher/consumer机制)中来介绍

2) 把监听器poll出来的消息通过这个dispatcher传到XXXManager中的具体处理方法上

关于connectionpool不详细介绍了，它是与具体Transport的连接池。

消息发出端[Client端]孤单寂寞与被遗弃感是最可怕的贫穷