RPC-server的创建过程

可以参考下面的这个链接,讲的非常清楚啦,一共五篇

http://blog.csdn.net/gj19890923/article/details/50407782


下面是转载的。

       在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: 这个慨念很简单, 就是函数, 即远程方法调用中的方法;
  • API version: 也就是server上提供的RPC api接口集合的版本号,openstack中1.0起步, servers可以一次提供多种api version,client每次请求时只需描述它所需要的最低version就ok;
  • Transport: 可以理解为传输载体,这个很好理解, 就是我们使用的消息队列中间件RabbitMQ, Qpid, ZeroMQ等等, 是负责整个消息处理的系统, 它负责消息传输直到提供给clients返回, 使用此系统者, 不用了解细节,  Openstack中实现的主要有这三种, AMQP标准下的rabbitMQ和Qpid, 和非AMQP的ZeroMQ, ZeroMQ更底层, 速度更快, 据说快10倍。
  • Target这是个很重要的概念, 它描述了信息的处理方式, 该发哪里去(server属性)和消息处理端(server)监听什么信息(topic 属性)。以下是Target的属性
exchange (defaults to CONF.control_exchange)
topic
server (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上run

3.存在多个接口版本的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

[python] view plain copy
 在CODE上查看代码片派生到我的代码片
  1. def listen(self, target):  
  2.         conn = self._get_connection(pooled=False)  
  3.   
  4.         listener = AMQPListener(self, conn)  
  5.   
  6.         conn.declare_topic_consumer(target.topic, listener)  
  7.         conn.declare_topic_consumer('%s.%s' % (target.topic, target.server),  
  8.                                     listener)  
  9.         conn.declare_fanout_consumer(target.topic, listener)  
  10.   
  11.         return listener  
以上代码可以看到listener里做了什么, declare了三个consumer, topic, topic.host, fanout, 正好呼应上面介绍的三种user case。关于consumer, 将会在另一遍博文(介绍amqp的publisher/consumer机制)中来介绍

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

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

       

消息发出端[Client端]

主要都写在rpcapi.py 中, 本文以Cinder项目为例,讲述delete_volume消息发出的过程。

[python] view plain copy
 在CODE上查看代码片派生到我的代码片
  1. class VolumeAPI(object):  
  2.     BASE_RPC_API_VERSION = '1.0'  
  3.   
  4.     def __init__(self, topic=None):  
  5.         super(VolumeAPI, self).__init__()  
  6.         target = messaging.Target(topic=CONF.volume_topic,  
  7.                                   version=self.BASE_RPC_API_VERSION)  
  8.         self.client = rpc.get_client(target, '1.15')  
  9.     ....  
  10.   
  11.     def delete_volume(self, ctxt, volume, unmanage_only=False):  
  12.         cctxt = self.client.prepare(server=volume['host'], version='1.15')  
  13.         cctxt.cast(ctxt, 'delete_volume',  
  14.                    volume_id=volume['id'],  
  15.                    unmanage_only=unmanage_only)  

RPCClient

            负责通过消息队列发消息到消息队列中,两种方法调用的方式:cast  & call。关于这两种方法也将会在另一遍博文(介绍amqp的publisher/consumer机制)中来介绍

            会利用Transport的send方法。



原文地址:https://www.cnblogs.com/double12gzh/p/10166171.html