OpenStack RPC框架浅析

def main():
    # 调用nova/service.py文件的Service类的create类方法实例化一个service类
    server = service.Service.create(binary=‘nova-compute‘,
                                    topic=CONF.compute_topic)
    service.serve(server)
service.wait()

def serve(server, workers=None):
    global _launcher
    if _launcher:
        raise RuntimeError(_(‘serve() can only be called once‘))

    # 这里的service是指oslo_service包导入的service了
    # 调用到oslo_service/service.py的launch方法
    _launcher = service.launch(CONF, server, workers=workers)

def launch(conf, service, workers=1, restart_method=‘reload‘):

    if workers is not None and workers <= 0:
        raise ValueError(_("Number of workers should be positive!"))

    # 默认传入的是None
    if workers is None or workers == 1:
        # 这里是初始化一个继承了Launcher类的ServiceLauncher类实例
        launcher = ServiceLauncher(conf, restart_method=restart_method)
    else:
        launcher = ProcessLauncher(conf, restart_method=restart_method)
    # 调用Launcher类里的launch_service方法，launch_service方法运行给定的service
    launcher.launch_service(service, workers=workers)

    return launcher

def launch_service(self, service, workers=1):
    if workers is not None and workers != 1:
        raise ValueError(_("Launcher asked to start multiple workers"))
    _check_service_base(service)
    service.backdoor_port = self.backdoor_port
    # 调用Services类的add方法来运行给定service
    # 其实最后也就是开辟了个绿色线程池和获取一个绿色线程运行service的start方法
    self.services.add(service)

def start(self):
    ......

    # 初始化oslo_messaging/target.py的Target类
    target = messaging.Target(topic=self.topic, server=self.host)

    endpoints = [
        self.manager,
        baserpc.BaseRPCAPI(self.manager.service_name,         
                self.backdoor_port)
    ]
    endpoints.extend(self.manager.additional_endpoints)

    # 获取nova/objects/base.py中的NovaObjectSerializer类实例
    # 用来序列化nova服务中的对象
    serializer = objects_base.NovaObjectSerializer()

    # 获取一个oslo_messaging/rpc/RPCServer类实例
    self.rpcserver = rpc.get_server(target, endpoints, serializer)
    # 调用到oslo_messaging/server.py的MessageHandlingServer类的start方法
    self.rpcserver.start()

    ......

def get_server(target, endpoints, serializer=None):

    # TRANSPORT一个transport类对象，里面包含发消息的driver实现对象，如果是rabbit则对应到实现rabbit的driver类
    # 获取TRANSPORT对象：<class ‘oslo_messaging.transport.Transport‘>
    # 更重要的是transport对象里的driver对象：   oslo_messaging._drivers.impl_rabbit.RabbitDriver
    assert TRANSPORT is not None

    if profiler:
        serializer = ProfilerRequestContextSerializer(serializer)
    else:
        serializer = RequestContextSerializer(serializer)

    # get_rpc_server在oslo_messaging/rpc/server.py文件中
    # 获取一个RPCServer实例
    return messaging.get_rpc_server(TRANSPORT,
                                    target,
                                    endpoints,
                                    executor=‘eventlet‘,
                                    serializer=serializer)

def get_rpc_server(transport, target, endpoints,
                   executor=‘blocking‘, serializer=None, access_policy=None):
    # 获取一个消息调度员，它能识别收到的消息的结构
    # 用于接收到消息后进行消息分发处理 A message dispatcher which understands RPC messages
    # oslo_messaging/rpc/dispatcher.py类的RPCDispatcher类
    # 解析消息然后调用相对应的方法进行处理
    dispatcher = rpc_dispatcher.RPCDispatcher(endpoints, serializer,
                                              access_policy)
    # 该类有个关键函数_process_incoming是在接收到消息时进行回调的
    return RPCServer(transport, target, dispatcher, executor)

def start(self, override_pool_size=None):
    ......

    try:
        # 这里程序调用到的是oslo_messaging/rpc/server.py的RPCServer类的_create_listener方法
        self.listener = self._create_listener()
    except driver_base.TransportDriverError as ex:
        raise ServerListenError(self.target, ex)
    ......

    self.listener.start(self._on_incoming)

def _create_listener(self):
    # oslo_messaging/transport/Transport类_listen方法
    return self.transport._listen(self._target, 1, None)

def _listen(self, target, batch_size, batch_timeout):
    if not (target.topic and target.server):
        raise exceptions.InvalidTarget(‘A server\‘s target must have ‘
                                       ‘topic and server names specified‘,
                                       target)
    # 这个_driver对应的是oslo_messaging._drivers.impl_rabbit.RabbitDriver
    return self._driver.listen(target, batch_size,
                               batch_timeout)

def listen(self, target, batch_size, batch_timeout):
    # 这里是从连接池里获取一个连接对象
    # conn是oslo_messaging/_drivers/impl_rabbit.py的类Connection实例
    conn = self._get_connection(rpc_common.PURPOSE_LISTEN)

    # 这个listen很关键，它被绑定为消费者的回调对象，也就是收到消息时是调用该对象，该对象实现了__call__方法
    # 所以可直接调用
    listener = AMQPListener(self, conn)

    conn.declare_topic_consumer(exchange_name=self._get_exchange(target),
                                topic=target.topic,
                                callback=listener)
    conn.declare_topic_consumer(exchange_name=self._get_exchange(target),
                                topic=‘%s.%s‘ % (target.topic,
                                                 target.server),
                                callback=listener)
    conn.declare_fanout_consumer(target.topic, listener)

    # 返回一个实现poll模式监听消息到来的类
    return base.PollStyleListenerAdapter(listener, batch_size,
                                         batch_timeout)

def start(self, on_incoming_callback):
    super(PollStyleListenerAdapter, self).start(on_incoming_callback)
    self._started = True
    # _listen_thread在__init__方法中定义了，如下行所示
    # self._listen_thread = threading.Thread(target=self._runner)
    # 所以是开启一个线程运行_runner函数
    self._listen_thread.start()

def _runner(self):
    while self._started:
        # 这里poll调用oslo_messaging/_drivers/amqpdriver.py的AMQPListener类poll函数
        incoming = self._poll_style_listener.poll(
            batch_size=self.batch_size, batch_timeout=self.batch_timeout)

        # 读到有消息，调用回调函数进行处理
        if incoming:
            # 该on_incoming_callback是oslo_messaging/server.py中self.listener.start(self._on_incoming)中传进来的
            # 是该文件中MessageHandlingServer类的_on_incoming函数
            # _on_incoming函数又调用到了RPCServer中的_process_incoming函数
            self.on_incoming_callback(incoming)

def poll(self, timeout=None):
    stopwatch = timeutils.StopWatch(duration=timeout).start()

    while not self._shutdown.is_set():
        self._message_operations_handler.process()

        if self.incoming:
            # 从incoming列表中获取第一个消息返回
            return self.incoming.pop(0)

        left = stopwatch.leftover(return_none=True)
        if left is None:
            left = self._current_timeout
        if left <= 0:
            return None

        try:
            # 获取所有队列的消息
            # oslo_messaging/dr_drivers/impl_rabbit.py的Connection类的consume函数
            # 将获取到的消息经过解析存到incoming列表中
            self.conn.consume(timeout=min(self._current_timeout, left))
        except rpc_common.Timeout:
            self._current_timeout = max(self._current_timeout * 2,
                                        ACK_REQUEUE_EVERY_SECONDS_MAX)
        else:
            self._current_timeout = ACK_REQUEUE_EVERY_SECONDS_MIN

    # NOTE(sileht): listener is stopped, just processes remaining messages
    # and operations
    self._message_operations_handler.process()
    if self.incoming:
        return self.incoming.pop(0)

    self._shutoff.set()

self.connection.drain_events(timeout=poll_timeout)

def drain_events(self, timeout=None):
    """Wait for an event on a channel."""
    # 等待事件通知
    chanmap = self.channels
    chanid, method_sig, args, content = self._wait_multiple(
        chanmap, None, timeout=timeout,
    )

    channel = chanmap[chanid]

    if (content and
            channel.auto_decode and
            hasattr(content, ‘content_encoding‘)):
        try:
            content.body = content.body.decode(content.content_encoding)
        except Exception:
            pass

    amqp_method = (self._method_override.get(method_sig) or
                   channel._METHOD_MAP.get(method_sig, None))

    if amqp_method is None:
        raise AMQPNotImplementedError(
            ‘Unknown AMQP method {0!r}‘.format(method_sig))

    if content is None:
        return amqp_method(channel, args)
    else:
        return amqp_method(channel, args, content)

listener = AMQPListener(self, conn)
conn.declare_topic_consumer(exchange_name=self._get_exchange(target),
                            topic=target.topic,
                            callback=listener)

def __call__(self, message):
    # 收到消息后解析消息结构体并构建成AMQPIncomingMessage结构
    # type(message)对应类：<class ‘oslo_messaging._drivers.impl_rabbit.RabbitMessage‘>
    ctxt = rpc_amqp.unpack_context(message)
    unique_id = self.msg_id_cache.check_duplicate_message(message)
    if ctxt.msg_id:
        LOG.debug("received message msg_id: %(msg_id)s reply to "
                  "%(queue)s", {‘queue‘: ctxt.reply_q,
                                ‘msg_id‘: ctxt.msg_id})
    else:
        LOG.debug("received message with unique_id: %s", unique_id)

    self.incoming.append(AMQPIncomingMessage(
        self,
        ctxt.to_dict(),
        message,
        unique_id,
        ctxt.msg_id,
        ctxt.reply_q,
        self._obsolete_reply_queues,
        self._message_operations_handler))

def _runner(self):
    while self._started:
        # 这里poll调用oslo_messaging/_drivers/amqpdriver.py的AMQPListener类poll函数
        incoming = self._poll_style_listener.poll(
            batch_size=self.batch_size, batch_timeout=self.batch_timeout)

        # 读到有消息，调用回调函数进行处理
        if incoming:
            # 该on_incoming_callback是oslo_messaging/server.py中self.listener.start(self._on_incoming)中传进来的
            # 是该文件中MessageHandlingServer类的_on_incoming函数
            # _on_incoming函数又调用到了RPCServer中的_process_incoming函数
            self.on_incoming_callback(incoming)

def _process_incoming(self, incoming):
    message = incoming[0]
    try:
        # 这里进行了消息确认发送
        # 会调用到kombu/message.py的Message类的ack函数
        # 表示该消息已经进行消费了，队列中可以删除该消息了
        message.acknowledge()
    except Exception:
        LOG.exception(_LE("Can not acknowledge message. Skip processing"))
        return

    failure = None
    try:
        # 调用oslo_messaging/rpc/dispatcher.py的RPCDispatcher类来处理消息
        # 该类的职责是找到消息对应的方法并执行
        res = self.dispatcher.dispatch(message)
    except rpc_dispatcher.ExpectedException as e:
        failure = e.exc_info
        LOG.debug(u‘Expected exception during message handling (%s)‘, e)
    except Exception:
        # current sys.exc_info() content can be overridden
        # by another exception raised by a log handler during
        # LOG.exception(). So keep a copy and delete it later.
        failure = sys.exc_info()
        LOG.exception(_LE(‘Exception during message handling‘))

    try:
        # 将执行结果返回
        if failure is None:
            message.reply(res)
        else:
            message.reply(failure=failure)
    except Exception:
        LOG.exception(_LE("Can not send reply for message"))
    finally:
            # NOTE(dhellmann): Remove circular object reference
            # between the current stack frame and the traceback in
            # exc_info.
            del failure

def dispatch(self, incoming):
    """Dispatch an RPC message to the appropriate endpoint method.

    :param incoming: incoming message
    :type incoming: IncomingMessage
    :raises: NoSuchMethod, UnsupportedVersion
    """
    message = incoming.message
    ctxt = incoming.ctxt

    method = message.get(‘method‘)
    args = message.get(‘args‘, {})
    namespace = message.get(‘namespace‘)
    version = message.get(‘version‘, ‘1.0‘)

    found_compatible = False
    # endpoints值是两个类
    # [<nova.compute.manager.ComputeManager object at 0x7f6eef157dd0>, <nova.baserpc.BaseRPCAPI object at 0x7f6ee4724c90>]
    # 从类中查找出方法进行调用
    for endpoint in self.endpoints:
        target = getattr(endpoint, ‘target‘, None)
        if not target:
            target = self._default_target

        if not (self._is_namespace(target, namespace) and
                self._is_compatible(target, version)):
            continue

        if hasattr(endpoint, method):
            if self.access_policy.is_allowed(endpoint, method):
                return self._do_dispatch(endpoint, method, ctxt, args)

        found_compatible = True

    if found_compatible:
        raise NoSuchMethod(method)
    else:
        raise UnsupportedVersion(version, method=method)

def _do_dispatch(self, endpoint, method, ctxt, args):
    ctxt = self.serializer.deserialize_context(ctxt)
    new_args = dict()
    for argname, arg in args.items():
        new_args[argname] = self.serializer.deserialize_entity(ctxt, arg)
    func = getattr(endpoint, method)
    # 调用方法
    result = func(ctxt, **new_args)
    return self.serializer.serialize_entity(ctxt, result)

def _start_server(self, req, id, body):
    .....
    try:
        # nova/compute/api.py
        self.compute_api.start(context, instance)
    except (exception.InstanceNotReady, exception.InstanceIsLocked) as e:
        .....

def start(self, context, instance):
    .....
    instance.task_state = task_states.POWERING_ON
    instance.save(expected_task_state=[None])
    self._record_action_start(context, instance, instance_actions.START)
    self.compute_rpcapi.start_instance(context, instance)
    .....

def start_instance(self, ctxt, instance):
    version = ‘4.0‘
    # self.router.by_instance(ctxt, instance)获取oslo_messaging/rpc/client.py的RPCClient类实例
    # prepare方法是用于设置一些属性并生成一个oslo_messaging/rpc/client.py的_CallContext类实例
    cctxt = self.router.by_instance(ctxt, instance).prepare(
            server=_compute_host(None, instance), version=version)
    # 该call方法是调用到oslo_messaging/rpc/client.py的_BaseCallContext(_CallContext的父类)类的call方法
    return cctxt.call(ctxt, ‘start_instance‘, instance=instance)

def call(self, ctxt, method, **kwargs):
    """Invoke a method and wait for a reply. See RPCClient.call()."""
    if self.target.fanout:
        raise exceptions.InvalidTarget(‘A call cannot be used with fanout‘,
                                       self.target)

    # 生成一个msg
    msg = self._make_message(ctxt, method, kwargs)
    # 序列化ctxt
    msg_ctxt = self.serializer.serialize_context(ctxt)

    timeout = self.timeout
    if self.timeout is None:
        timeout = self.conf.rpc_response_timeout

    self._check_version_cap(msg.get(‘version‘))

    try:
        # 调用transport中的_send方法
        # _send方法中又是使用_driver对象调用send方法
        # _driver对象是oslo_messaging._drivers.impl_rabbit.RabbitDriver
        # 所以是调用RabbitDriver类里的send方法，实际是调用它父类AMQPDriverBase的send方法
        result = self.transport._send(self.target, msg_ctxt, msg,
                                      wait_for_reply=True, timeout=timeout,
                                      retry=self.retry)
    except driver_base.TransportDriverError as ex:
        raise ClientSendError(self.target, ex)

    return self.serializer.deserialize_entity(ctxt, result)

def _send(self, target, ctxt, message,
          wait_for_reply=None, timeout=None,
          envelope=True, notify=False, retry=None):

    msg = message

    if wait_for_reply:
        msg_id = uuid.uuid4().hex
        msg.update({‘_msg_id‘: msg_id})
        # _get_reply_q方法获取一个ReplyWaiter对象，开始poll模式等待获取返回消息
        # 同时使用waiters(ReplyWaiter实例对象)管理返回的消息
        msg.update({‘_reply_q‘: self._get_reply_q()})

    # 获取一个唯一uuid添加到msg字典中
    rpc_amqp._add_unique_id(msg)
    unique_id = msg[rpc_amqp.UNIQUE_ID]

    # 把ctxt字典值也更新到msg中
    rpc_amqp.pack_context(msg, ctxt)

    if envelope:
        msg = rpc_common.serialize_msg(msg)

    if wait_for_reply:
        # 把该消息加到waiters里去监控管理
        self._waiter.listen(msg_id)
        log_msg = "CALL msg_id: %s " % msg_id
    else:
        log_msg = "CAST unique_id: %s " % unique_id

    try:
        # 根据target保存的发送模式发送消息
        with self._get_connection(rpc_common.PURPOSE_SEND) as conn:
            if notify:
                exchange = self._get_exchange(target)
                log_msg += "NOTIFY exchange ‘%(exchange)s‘"                            " topic ‘%(topic)s‘" % {
                               ‘exchange‘: exchange,
                               ‘topic‘: target.topic}
                LOG.debug(log_msg)
                conn.notify_send(exchange, target.topic, msg, retry=retry)
            elif target.fanout:
                log_msg += "FANOUT topic ‘%(topic)s‘" % {
                    ‘topic‘: target.topic}
                LOG.debug(log_msg)
                conn.fanout_send(target.topic, msg, retry=retry)
            else:
                topic = target.topic
                exchange = self._get_exchange(target)
                if target.server:
                    topic = ‘%s.%s‘ % (target.topic, target.server)
                log_msg += "exchange ‘%(exchange)s‘"                            " topic ‘%(topic)s‘" % {
                               ‘exchange‘: exchange,
                               ‘topic‘: topic}
                LOG.debug(log_msg)
                # 发送topic模式的队列
                # 调用oslo_messaging/_drivers/impl_rabbit.py中Connection类实例的topic_send方法
                conn.topic_send(exchange_name=exchange, topic=topic,
                                msg=msg, timeout=timeout, retry=retry)

        if wait_for_reply:
            # 等待返回或消息超时返回
            # 轮询方式检测消息有没有被返回并放到对应的字典中
            result = self._waiter.wait(msg_id, timeout)
            if isinstance(result, Exception):
                raise result
            return result
    finally:
        if wait_for_reply:
            self._waiter.unlisten(msg_id) 

def consume(self, timeout=None):
    """Consume from all queues/consumers."""

    timer = rpc_common.DecayingTimer(duration=timeout)
    timer.start()

    def _raise_timeout(exc):
        LOG.debug(‘Timed out waiting for RPC response: %s‘, exc)
        raise rpc_common.Timeout()

    def _recoverable_error_callback(exc):
        # 判断异常类型是不是非Timeout类型，因为Timeout类型是由drain_events函数获取
        # 消息等待超时导致的，属于正常的异常，除了这种异常，其它异常则都会被视作需要重建消费者
        if not isinstance(exc, rpc_common.Timeout):
            self._new_tags = set(self._consumers.values())
        timer.check_return(_raise_timeout, exc)

    def _error_callback(exc):
        # 将异常交给_recoverable_error_callback函数进行处理
        _recoverable_error_callback(exc)
        LOG.error(_LE(‘Failed to consume message from queue: %s‘),
                  exc)

    def _consume():
        # NOTE(sileht): in case the acknowledgment or requeue of a
        # message fail, the kombu transport can be disconnected
        # In this case, we must redeclare our consumers, so raise
        # a recoverable error to trigger the reconnection code.
        # 这里是判断了连接是否还正常，如果不正常，我们需要重新获取连接并且重新定义consumer
        if not self.connection.connected:
            # 这里抛错以进入重连机制
            raise self.connection.recoverable_connection_errors[0]

        while self._new_tags:
            for consumer, tag in self._consumers.items():
                if tag in self._new_tags:
                    # 如果是新标签则消费者也需建立
                    # 在重建channel时这里就是重建消费者了
                    # 这里consumer是该文件的Consumer类实例，该consume函数会调用到kombu中的consume函数定义消费者
                    consumer.consume(self, tag=tag)
                    self._new_tags.remove(tag)

        poll_timeout = (self._poll_timeout if timeout is None
                        else min(timeout, self._poll_timeout))
        while True:
            if self._consume_loop_stopped:
                return

            if self._heartbeat_supported_and_enabled():
                # 心跳检查，如果连不通则抛错
                # 抛错则会在kombu中进行重连机制
                self._heartbeat_check()

            try:
                # 调用kombu/connection.py的Connection类的drain_events方法，等待来自服务器的单个事件，所以这是事件触发型的
                # 其中里面的supports_librabbitmq()=False(因为环境支持’eventlet’，所以未采用’default’，所以返回False
                # 最后是调用到kombu/transport/pyamqp.py的drain_events方法
                # 再调用到amqp包的Connection类的drain_events(amqp/connection.py)
                self.connection.drain_events(timeout=poll_timeout)
                return
            except socket.timeout as exc:
                # 超时会进入这个逻辑，check_return会raise一个Exception，从而导致ensure中抛异常被捕获调用了error_callback函数
                # error_callback函数又调用了recoverable_error_callback函数
                # 从而导致日志中经常可以看到_recoverable_error_callback
                poll_timeout = timer.check_return(
                    _raise_timeout, exc, maximum=self._poll_timeout)

    with self._connection_lock:
        self.ensure(_consume,
                    recoverable_error_callback=_recoverable_error_callback,
                    error_callback=_error_callback)

def ensure(self, method, retry=None,
           recoverable_error_callback=None, error_callback=None,
           timeout_is_error=True):
    .....

    # 在kombu中如果进入了异常重连处理机制会回调该函数
    def on_error(exc, interval):
        LOG.debug("[%s] Received recoverable error from kombu:"
                  % self.connection_id,
                  exc_info=True)

        # 执行_recoverable_error_callback函数处理异常
        recoverable_error_callback and recoverable_error_callback(exc)

        interval = (self.kombu_reconnect_delay + interval
                    if self.kombu_reconnect_delay > 0
                    else interval)

        info = {‘err_str‘: exc, ‘sleep_time‘: interval}
        info.update(self._get_connection_info())

        if ‘Socket closed‘ in six.text_type(exc):
            LOG.error(_LE(‘[%(connection_id)s] AMQP server‘
                          ‘ %(hostname)s:%(port)s closed‘
                          ‘ the connection. Check login credentials:‘
                          ‘ %(err_str)s‘), info)
        else:
            LOG.error(_LE(‘[%(connection_id)s] AMQP server on ‘
                          ‘%(hostname)s:%(port)s is unreachable: ‘
                          ‘%(err_str)s. Trying again in ‘
                          ‘%(sleep_time)d seconds. Client port: ‘
                          ‘%(client_port)s‘), info)

    ......

    # 当在kombu中执行autoretry时抛出异常了并在异常处理时重新连接了其它节点
    # 则会回调该函数
    def on_reconnection(new_channel):
        # 更新channel
        self._set_current_channel(new_channel)
        self.set_transport_socket_timeout()

    def execute_method(channel):
        # 更新channel
        self._set_current_channel(channel)
        # 这个method指的就是_consume函数
        # 注意我这里指的是consume调入的时候该method就是_consume函数
        # 因为该ensure函数是很多函数都会调用的，每个函数都会传入它的method函数进行回调
        # 我这里是为了方便理解就这样指明了,文中还有很多地方也是如此指明，就不一一解释了
        method()

    # NOTE(sileht): Some dummy driver like the in-memory one doesn‘t
    # have notion of recoverable connection, so we must raise the original
    # exception like kombu does in this case.
    has_modern_errors = hasattr(
        self.connection.transport, ‘recoverable_connection_errors‘,
    )
    if has_modern_errors:
        recoverable_errors = (
            self.connection.recoverable_channel_errors +
            self.connection.recoverable_connection_errors)
    else:
        recoverable_errors = ()

    try:
        autoretry_method = self.connection.autoretry(
            execute_method, channel=self.channel,
            max_retries=retry,
            errback=on_error,
            interval_start=self.interval_start or 1,
            interval_step=self.interval_stepping,
            interval_max=self.interval_max,
            on_revive=on_reconnection)
        ret, channel = autoretry_method()
        self._set_current_channel(channel)
        return ret
    except recoverable_errors as exc:
        LOG.debug("Received recoverable error from kombu:",
                  exc_info=True)
        # 在kombu重建立连接失败时会跑入该逻辑，调用error_callback
        # 如果是consume函数调入该函数的话，则该函数是_error_callback函数
        error_callback and error_callback(exc)
        self._set_current_channel(None)
        # NOTE(sileht): number of retry exceeded and the connection
        # is still broken
        info = {‘err_str‘: exc, ‘retry‘: retry}
        info.update(self.connection.info())
        msg = _(‘Unable to connect to AMQP server on ‘
                ‘%(hostname)s:%(port)s after %(retry)s ‘
                ‘tries: %(err_str)s‘) % info
        LOG.error(msg)
        raise exceptions.MessageDeliveryFailure(msg)
    except rpc_amqp.AMQPDestinationNotFound:
        # NOTE(sileht): we must reraise this without
        # trigger error_callback
        raise
    except Exception as exc:
        error_callback and error_callback(exc)
        Raise

def autoretry(self, fun, channel=None, **ensure_options):
    channels = [channel]
    create_channel = self.channel

    class Revival(object):
        __name__ = getattr(fun, ‘__name__‘, None)
        __module__ = getattr(fun, ‘__module__‘, None)
        __doc__ = getattr(fun, ‘__doc__‘, None)

        def revive(self, channel):
            channels[0] = channel

        def __call__(self, *args, **kwargs):
            if channels[0] is None:
                self.revive(create_channel())
            # 执行oslo_messaging中impl_rabbit的ensure中的execute_method函数
            # execute_method最终又是回调到_consume函数
            return fun(*args, channel=channels[0], **kwargs), channels[0]

    revive = Revival()
    # 返回了一个_ensure闭包函数
    # 但oslo_messaging的ensure函数中的下一行便是执行该闭包函数
    return self.ensure(revive, revive, **ensure_options)

def ensure(self, obj, fun, errback=None, max_retries=None,
           interval_start=1, interval_step=1, interval_max=1,
           on_revive=None):
           
    def _ensured(*args, **kwargs):
        got_connection = 0
        conn_errors = self.recoverable_connection_errors
        chan_errors = self.recoverable_channel_errors
        has_modern_errors = hasattr(
            self.transport, ‘recoverable_connection_errors‘,
        )
        for retries in count(0):  # for infinity
            try:
                # 调用了Revival类的__call__函数
                # 在进行的一系列调用中如果有异常抛出则进入下面的重连机制
                return fun(*args, **kwargs)
            except conn_errors as exc:
                if got_connection and not has_modern_errors:
                    raise
                if max_retries is not None and retries > max_retries:
                    raise
                self._debug(‘ensure connection error: %r‘, exc, exc_info=1)
                self._connection = None
                self._do_close_self()
                errback and errback(exc, 0)
                remaining_retries = None
                if max_retries is not None:
                    remaining_retries = max(max_retries - retries, 1)
                # 尝试重新建立连接，确保有连接建立成功
                self.ensure_connection(errback,
                                       remaining_retries,
                                       interval_start,
                                       interval_step,
                                       interval_max)
                # 在连接上获取新的channel
                new_channel = self.channel()
                self.revive(new_channel)
                obj.revive(new_channel)
                if on_revive:
                    # 调用oslo_messaging中的on_reconnection函数
                    # 将获得的新channel赋给Connection类的channel
                    on_revive(new_channel)
                got_connection += 1
            except chan_errors as exc:
                if max_retries is not None and retries > max_retries:
                    raise
                self._debug(‘ensure channel error: %r‘, exc, exc_info=1)
                errback and errback(exc, 0)
    _ensured.__name__ = "%s(ensured)" % fun.__name__
    _ensured.__doc__ = fun.__doc__
    _ensured.__module__ = fun.__module__
    return _ensured