Oslo系列之oslo.messaging

生命的道路上永远没有捷径可言，只有脚踏实地走下去。

今天介绍一下 openstack 中关于 RabbitMq 和 oslo_messaging 库之间的前生今世

一定要弄清楚的一个问题就是：直接使用rabbitmq和使用oslo_messaging这个库间接的使用有什么区别。

olso_messaging实际上是在rabbitmq的基础上通过一些列的调用，最终暴露给用户一个简单的使用接口，用户不必关心内部的实现，只用配置好配置文件，进行简单的函数调用即可。
并且由于这个库是openstack的标准库，里面的一些函数命名和默认参数都是针对于openstack的概念来讲的。

基本结构就是：

openstack -> oslo_messaging -> kombu -> AMQP -> socket

首先看下 oslo_messaging 中对象封装：

概念简介

Transport

Transport(传输层)主要实现RPC底层的通信(比如socket)以及事件循环,多线程等其他功能.可以通过URL来获得不同transport的句柄.URL的格式为:

transport://user:password@host:port[,hostN:portN]/virtual_host

目前支持的Transport有rabbit,qpid与zmq,分别对应不同的后端消息总线.用户可以使用oslo.messaging.get_transport函数来获得transport对象实例的句柄.

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

Target

Target封装了指定某一个消息最终目的地的所有信息,下表所示为其所具有的属性:

参数=默认值	说明
exchange = None	(字符串类型)topic所属的范围,不指定的话默认使用配置文件中的control_exchange选项
topic = None	(字符串类型)一个topic可以用来标识服务器所暴露的一组接口(一个接口包含多个可被远程调用的方法).允许多个服务器暴露同一组接口,消息会以轮循的方式发送给多个服务器中的某一个
namespace = None	(字符串类型)用来标识服务器所暴露的某个特定接口(多个可被远程调用的方法)
version = None	(字符串类型)服务器所暴露的接口支持M.N类型的版本号.次版本号(N)的增加表示新的接口向前兼容,主版本号(M)的增加表示新接口和旧接口不兼容.RPC服务器可以实现多个不同的主版本号接口.
server = None	(字符串类型)客户端可以指定此参数来要求消息的目的地是某个特定的服务器,而不是一组同属某个topic的服务器中的任意一台.
fanout = None	(布尔型)当设置为真时,消息会被发送到同属某个topic的所有服务器上,而不是其中的一台.

在不同的应用场景下,构造Target对象需要不同的参数:创建一个RPC服务器时,需要topic和server参数,exchange参数可选;指定一个endpoint时,namespace和version是可选的;客户端发送消息时,需要topic参数,其他可选.

Server

一个RPC服务器可以暴露多个endpoint,每个endpoint包含一组方法,这组方法是可以被客户端通过某种Transport对象远程调用的.创建Server对象时,需要指定Transport,Target和一组endpoint.

RPC Client

通过RPC Client,可以远程调用RPC Sever上的方法.远程调用时,需要提供一个字典对象来指明调用的上下文,调用方法的名字和传递给调用方法的参数(用字典表示).

有cast和call两种远程调用方式.通过cast方式远程调用,请求发送后就直接返回了;通过call方式调用,需要等响应从服务器返回.

Notifier

Notifier用来通过某种transport发送通知消息.通知消息遵循如下的格式:

import six
import uuid
from oslo_utils import timeutils
 
{'message_id': six.text_type(uuid.uuid4()), #消息id号
 'publisher_id': 'compute.hos1',    #发送者id
 'timestamp': timeutils.utcnow(),   #时间戳
 'priority': 'WARN',                #通知优先级
 'event_type': 'compute.create_instance',   #通知类型
 'payload': {'instance_id': 12, ...}}       #通知内容

可以在不同的优先级别上发送通知,这些优先级包括sample,critical,error,warn,info,debug,audit等.

Notification Listener

Notification Listener和Server类似,一个Notification Listener对象可以暴露多个endpoint,每个endpoint包含一组方法.但是与Server对象中的endpoint不同的是,这里的endpoint中的方法对应通知消息的不同优先级.比如：

import oslo_messaging
 
class ErrorEndpoint:
    def error(self, ctxt, publisher_id, event_type, payload, metadata):
        do_something(payload)
        return oslo_messaging.NotificationResult.HANDLED

endpoint中的方法如果返回messaging.NotificationResult.HANDLED或者None,表示这个通知消息已经确认被处理;如果返回messaging.NotificationResult.REQUEUE,表示这个通知消息要重新进入消息队列.

下面是一个利用oslo_messaging来实现远程过程调用的示例.

from oslo_config import cfg
import oslo_messaging as messaging
 
class ServerControlEndpoint(object):
    target = messaging.Target(namespace='controle',
                              version='2.0')
 
    def __init__(self, server):
        self.server = server
 
    def stop(self, ctx):
        if self.server:
            self.server.stop()
 
class TestEndpoint(object):
    def test(self, ctx, arg):
        return arg
 
 
transport = messaging.get_transport(cfg.CONF)
target = messaging.Target(topic='test',
                          server='server1')
endpoints = [
    ServerControlEndpoint(None),
    TestEndpoint(),
]
 
server = messaging.get_rpc_server(transport, target, endpoints,
                                  executor='blocking')
server.start()
server.wait()

这个例子里,定义了两个不同的endpoint:ServerControlEndpoint与TestEndpoint.这两个endpoint中的方法stop和test都可以被客户端远程调用.

创建rpc server对象之前,需要先创建transport和target对象,这里使用get_transport()函数来获得transport对象的句柄,get_transport()的参数如下表所示:

参数=默认值	说明
conf	(oslo.config.cfg.ConfigOpts类型)oslo.config配置项对象
url = None	(字符串或者oslo.messaging.Transport类型)transport URL.如果为空,采用conf配置中的transport_url项所指定的值
namespace = None	(字符串类型)用来标识服务器所暴露的某个特定接口(多个可被远程调用的方法)
allowed_remote_exmods = None	(列表类型)Python模块的列表.客户端可用列表里的模块来deserialize异常
aliases = None	(字典类型)transport别名和transport名称之间的对应关系

conf对象里,除了包含transport_url项外,还可以包含control_exchange项.control_exchange用来指明topic所属的默认范围,默认为”openstack”.可以使用oslo.messaging.set_transport_defaults()函数来修改默认值.

此处构建的Target对象是用来建立RPC Server的,所以需指明topic和server参数.用户定义的endpoint对象也可以包含一个target属性,用来指明这个endpoint所支持的特定的namespace和version.

这里使用get_rpc_server()函数创建server对象,然后调用server对象的start方法开始接收远程调用.get_rpc_server()函数的参数如下表所求:

参数=默认值	说明
transport	(Transpor类型)transport对象
target	(Target类型)target对象,用来指明监听的exchange,topic和server
endpoints	(列表类型)包含了endpoints对象实例的列表
executor=’blocking’	(字符串类型)用来指明消息接收和发收的方式:目前支持两种方式: blocking:在这种方式中,用户调用start函数后,在start函数中开始请求处理循环:用户线程阻塞,处理下一个请求.直到用户调用了stop函数后,这个处理循环才会退出.消息的接收和分发处理都在调用start函数的线程中完成. eventlet:在这种方式中,会有一个协程GreenThread来处理消息的接收,然后有其他不同的GreenThread来处理不同消息的分发处理.调用start的用户线程不会被阻塞
serializer = None	(Serializer类型)用来序列化/反序列化消息

#client.py 客户端
 
from oslo_config import cfg
import oslo_messaging as messaging
 
transport = messaging.get_transport(cfg.CONF)
target = messaging.Target(topic='test')
client = messaging.RPCClient(transport, target)
ret = client.call(ctxt={},
                  method='test',
                  arg='myarg')
 
cctx = client.prepare(namespace='control', version='2.0')
cctx.cast({}, 'stop')

这里target对象构造时,必须要有的参数只有topic,创建RPCClient对象时,可以接受的参数如下表所示:

参数=默认值	说明
transport	(Transport类型)transport对象
target	(Taget类型)该client对象的默认target对象
timeout = None	(整数或者浮点数类型)客户端调用call方法时超时时间(秒)
version_cap = None	(字符串类型)最大所支持的版本号.当版本号超过时,会扔出RPCVersionCapError异常
serializer = None	（Serializer类型)用来序列化/反序列化消息
retry = None	(整数类型)连接重试次数:None或者-1:一直重试0:不重试>0:重试次数

远程调用时,需要传入调用上下文,调用方法的名字和传给调用方法的参数.

Target对象的属性在RPCClient对象构造以后,还可以通过prepare()方法修改.可以修改的属性包括exchange,topic,namespace,version,server,fanout,timeout,version_cap和retry.

修改后的target属性只在这个prepare()方法返回的对象中有效.

下面我们再来看一个利用oslo_messaing实现通知消息处理的例子:

#notification_listener.py 消息通知处理
 
from oslo_config import cfg
import oslo_messaging as messaging
 
class NotificationEndPoint(object):
    def warn(self, ctxt, publisher_id, event_type, payload, metadata):
        do_something(payload)
 
class ErrorEndpoint(object):
    def error(self, ctxt, publisher_id, event_type, payload, metadata):
        do_something(payload)
 
 
transport = messaging.get_transport(cfg.CONF)
targets = [
    messaging.Target(topic='notifications'),
    messaging.Target(topic='notifications_bis')
]
endpoints = [
    NotificationEndPoint(),
    ErrorEndpoint(),
]
listener = messaging.get_notification_listener(transport,
                                               targets,
                                               endpoints)
 
listener.start()
listener.wait()

通知消息处理的endpoint对象和远程过程调用的endpoint对象不同,对象定义的方法要和通知消息的优先级一一对应.我们可以为每个endpoint指定所对应的target对象.

最后调用get_notificaton_listener()函数构造notification listener对象,get_notification_listener()函数的参数如下表所示:

参数=默认值	说明
transport	(Transport类型)transport对象
target	(列表类型)target对象的列表,用来指明endpoints列表中的每一个endpoint所侦听处理的exchange和topic
endpoints	(列表类型)包含了endpoints对象实例的列表
executor=’blocking’	(字符串类型)用来指明消息接收和发收的方式:目前支持两种方式: blocking:在这种方式中,用户调用start函数后,在start函数中开始请求处理循环:用户线程阻塞,处理下一个请求.直到用户调用了stop函数后,这个处理循环才会退出.消息的接收和分发处理都在调用start函数的线程中完成. eventlet:在这种方式中,会有一个协程GreenThread来处理消息的接收,然后有其他不同的GreenThread来处理不同消息的分发处理.调用start的用户线程不会被阻塞
serializer=None	（Serializer类型）用来序列化/反序列化消息
allow_requeue=False	(布尔类型)如果为真,表示支持NotificationResult.REQUEUE

相对应的发送消息通知的代码如下:

#notifier_send.py
 
from oslo_config import cfg
import oslo_messaging as messaging
 
transport = messaging.get_transport(cfg.CONF)
notifier = messaging.Notifier(transport,
                              driver='messaging',
                              topic='notifications')
 
notifier2 = notifier.prepare(publisher_id='compute')
notifier2.error(ctxt={},
                event_type='my_type',
                payload={'content': 'error occurred'})

1	发送通知消息时,首先要构造Notifier对象,此时可能需要指定的参数如下表所示:

参数=默认值	说明
transport	（Transport类型)transport对象
target	(列表类型)target对象的列表,用来指明endpoints列表中的每一个endpoint所侦听处理的exchange和topic
publish_id = None	(字符串类型)发送者id
driver = None	(字符串类型)后台驱动.一般采用”messaging”.如果没有指定,会使用配置文件中的notificaton_driver的值
topic = None	(字符串类型)发送消息的topic.如果没有指定,会使用配置文件中的notification_topics的值
serializer = None	（Serializer类型）用来序列化/反序列化消息

初始化Notifier对象的操作比较复杂,所以可以用prepare()方法修改已创建的Notifier对象,prepare()方法返回的是新的Notifier对象的实例.它的参数如下表所示:

参数 = 默认值	说明
publish_id = None	(字符串类型)发送者id
retry = None	(整数类型)连接重试次数:None或者-1:一直重试0:不重试>0:重试次数

最后可以调用Notifier对象的不同方法(error, critical, warn, 等等)发送不同优先级的消息通知.

源码分析

根据上个章节，我们可以看到其实这个库最终暴漏给用户的是两个概念：1.rpc，2.notification

下面我们来根据基本源码分析一下这两个概念

rpc

rpc(即远程调用)的概念被划分为调用方和被调用方
调用方称为client：rpc_client
被调用方称为server：rpc_server
使用时，被调用方server.start,等待调用方client.cast 或 clinet.call即可发起阻塞或非阻塞的远程调用。

当rpc client执行一次远程调用时实际发生了什么呢？
（代码在oslo_messaging/rpc/client.py文件里）

rpc client

首先构建 rpc client ，实例化 RPCClient

self.rpc_client = messaging.get_rpc_client(
    messaging.get_transport(),
    version='1.0'
)

def get_rpc_client(transport, retry=None, **kwargs):
    """Return a configured oslo_messaging RPCClient."""
    target = oslo_messaging.Target(**kwargs)
    serializer = oslo_serializer.RequestContextSerializer(
        oslo_serializer.JsonPayloadSerializer())
    return oslo_messaging.RPCClient(transport, target,
                                    serializer=serializer,
                                    retry=retry)

class RPCClient(_BaseCallContext):
      _marker = _BaseCallContext._marker

    def __init__(self, transport, target,
                 timeout=None, version_cap=None, serializer=None, retry=None,
                 call_monitor_timeout=None, transport_options=None):
        if serializer is None:
            serializer = msg_serializer.NoOpSerializer()

        if not isinstance(transport, msg_transport.RPCTransport):
            LOG.warning("Using notification transport for RPC. Please use "
                        "get_rpc_transport to obtain an RPC transport "
                        "instance.")

        super(RPCClient, self).__init__(
            transport, target, serializer, timeout, version_cap, retry,
            call_monitor_timeout, transport_options
        )

        self.conf.register_opts(_client_opts)

然后通过 call 或者 cast 调用

1	self.rpc_client.prepare(topic=topic).call(cxt, method, **args)

@six.add_metaclass(abc.ABCMeta)
class _BaseCallContext(object):

    _marker = object()

    def __init__(self, transport, target, serializer,
                 timeout=None, version_cap=None, retry=None,
                 call_monitor_timeout=None, transport_options=None):
        self.conf = transport.conf

        self.transport = transport
        self.target = target
        self.serializer = serializer
        self.timeout = timeout
        self.call_monitor_timeout = call_monitor_timeout
        self.retry = retry
        self.version_cap = version_cap
        self.transport_options = transport_options

        super(_BaseCallContext, self).__init__()
		def cast(self, ctxt, method, **kwargs):
        """Invoke a method and return immediately. See RPCClient.cast()."""
        msg = self._make_message(ctxt, method, kwargs)
        msg_ctxt = self.serializer.serialize_context(ctxt)

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

        try:
            self.transport._send(self.target, msg_ctxt, msg,
                                 retry=self.retry,
                                 transport_options=self.transport_options)
        except driver_base.TransportDriverError as ex:
            raise ClientSendError(self.target, ex)

    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 = self._make_message(ctxt, method, kwargs)
        msg_ctxt = self.serializer.serialize_context(ctxt)

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

        cm_timeout = self.call_monitor_timeout

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

        try:
            result = \
                self.transport._send(self.target, msg_ctxt, msg,
                                     wait_for_reply=True, timeout=timeout,
                                     call_monitor_timeout=cm_timeout,
                                     retry=self.retry,
                                     transport_options=self.transport_options)
        except driver_base.TransportDriverError as ex:
            raise ClientSendError(self.target, ex)

        return self.serializer.deserialize_entity(ctxt, result)

可以看到在 29 行和 54 行，两个方法都是执行了 transport._send

只有参数不同，这里最大的区别其实是wait_for_reply这个参数，顾名思义wait or no wait也就是我们说的阻塞/非阻塞。
那_send这个方法，最重要的两关键一个是transport本身，一个是target参数，这两个东西是rpc client init的时候必须要传的参数，

transport 参数是由（osllo_messaging/transport.py文件）_get_transport方法而来:

def _get_transport(conf, url=None, allowed_remote_exmods=None,
                   transport_cls=RPCTransport):
    allowed_remote_exmods = allowed_remote_exmods or []
    conf.register_opts(_transport_opts)

    if not isinstance(url, TransportURL):
        url = TransportURL.parse(conf, url)

    kwargs = dict(default_exchange=conf.control_exchange,
                  allowed_remote_exmods=allowed_remote_exmods)

    try:
        mgr = driver.DriverManager('oslo.messaging.drivers',
                                   url.transport.split('+')[0],
                                   invoke_on_load=True,
                                   invoke_args=[conf, url],
                                   invoke_kwds=kwargs)
    except RuntimeError as ex:
        raise DriverLoadFailure(url.transport, ex)

    return transport_cls(mgr.driver)

这里url是配置文件里配的，这里以rabbitmq为例
entry_point到oslo_messaging._drivers.impl_rabbit:RabbitDriver，最终获得到的是RabbitDriver的实例。

target 直接实例化即可，这里注意到两个参数exchange和topic，和rabbitmq里的exchange和routing_key的概念一致

那我们接着来看 transport._send方法，前面也说到了transport此时是RabbitDriver

RabbitDriver 继承自 AMQPDriverBase 继承自 BaseDriver

_send 方法在AMQPDriverBase中：

class AMQPDriverBase(base.BaseDriver):
    missing_destination_retry_timeout = 0

    def __init__(self, conf, url, connection_pool,
                 default_exchange=None, allowed_remote_exmods=None):
        super(AMQPDriverBase, self).__init__(conf, url, default_exchange,
                                             allowed_remote_exmods)

        self._default_exchange = default_exchange

        self._connection_pool = connection_pool

        self._reply_q_lock = threading.Lock()
        self._reply_q = None
        self._reply_q_conn = None
        self._waiter = None    
    def _send(self, target, ctxt, message,
              wait_for_reply=None, timeout=None, call_monitor_timeout=None,
              envelope=True, notify=False, retry=None, transport_options=None):

        msg = message

        if wait_for_reply:
            msg_id = uuid.uuid4().hex
            msg.update({'_msg_id': msg_id})
            msg.update({'_reply_q': self._get_reply_q()})
            msg.update({'_timeout': call_monitor_timeout})

        rpc_amqp._add_unique_id(msg)
        unique_id = msg[rpc_amqp.UNIQUE_ID]

        rpc_amqp.pack_context(msg, ctxt)

        if envelope:
            msg = rpc_common.serialize_msg(msg)

        if wait_for_reply:
            self._waiter.listen(msg_id)
            log_msg = "CALL msg_id: %s " % msg_id
        else:
            log_msg = "CAST unique_id: %s " % unique_id

        try:
            with self._get_connection(rpc_common.PURPOSE_SEND) as conn:
                if notify:
                    exchange = self._get_exchange(target)
                    LOG.debug(log_msg + "NOTIFY exchange '%(exchange)s'"
                              " topic '%(topic)s'", {'exchange': exchange,
                                                     'topic': target.topic})
                    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.debug(log_msg + "exchange '%(exchange)s'"
                              " topic '%(topic)s'", {'exchange': exchange,
                                                     'topic': topic})
                    conn.topic_send(exchange_name=exchange, topic=topic,
                                    msg=msg, timeout=timeout, retry=retry,
                                    transport_options=transport_options)

            if wait_for_reply:
                result = self._waiter.wait(msg_id, timeout,
                                           call_monitor_timeout)
                if isinstance(result, Exception):
                    raise result
                return result
        finally:
            if wait_for_reply:
                self._waiter.unlisten(msg_id)

我们看下 44 行到 66 行，回顾上面的cast和call函数里调用_send的时候是没有传notify找个参数的，所以第一个条件一定不成立

那看接下来的两个case，elif target.fanout/else（这里的fanout与rabbitmq本身的fanout意义是一样的）那也就是说我们在生成target或者client.prepare的时候可以通过指定fanout这个参数来决定进入哪个case，(注意第三个case里如果指定了target.server那么topic是target.topic和target.server二者相结合)那我们这里来看一下conn.fanout_send和conn.topic_send这两个方法(conn是enter exit getattr的产物，具体本文不细说了，这里只要知道最终调用到了oslo_messaging/_drivers/impl_rabbit.py里Connection这个类就可以了)：

conn.fanout_send

class Connection(object):
		def fanout_send(self, topic, msg, retry=None):
        """Send a 'fanout' message."""
        exchange = kombu.entity.Exchange(name='%s_fanout' % topic,
                                         type='fanout',
                                         durable=False,
                                         auto_delete=True)

        self._ensure_publishing(self._publish, exchange, msg, retry=retry)

其实到这里基本上就清楚了fanout_send就是往名叫target.topic + “_fanout”这个exchange里发送fanout模式的消息，所有bind到这个exchange的queue都会收到这条消息，如果这个exchange没有创建过，在self.publish方法里会被declare.

conn.topic_send

class Connection(object):
    def topic_send(self, exchange_name, topic, msg, timeout=None, retry=None,
                   transport_options=None):
        """Send a 'topic' message."""
        exchange = kombu.entity.Exchange(
            name=exchange_name,
            type='topic',
            durable=self.amqp_durable_queues,
            auto_delete=self.amqp_auto_delete)

        self._ensure_publishing(self._publish, exchange, msg,
                                routing_key=topic, timeout=timeout,
                                retry=retry,
                                transport_options=transport_options)

topic_send就是以topic做为routing_key 以exchange_name这个参数值命名的exchange里发送topic模式的消息，这里注意区别就是exchange_name是上级调用_get_exchange方法得来的

1
2
3

class Connection(object):
		def _get_exchange(self, target):
        return target.exchange or self._default_exchange

_default_exchange如果仔细看的话前面其实前面的截图里有，就是conf.control_exchange
默认是openstack（这里大概知道点为啥oslo_messaging是为openstack搞得了吧😄）,关键还是取决于target,如果target里没有指定才会用配置文件的。
同样，如果这个exchange没有创建过，在self.publish方法里会被declare.

rpc server

首先要获得一个rpc server的实例

def get_rpc_server(transport, target, endpoints,
                   executor='blocking', serializer=None, access_policy=None):
  	dispatcher = rpc_dispatcher.RPCDispatcher(endpoints, serializer,
                                              access_policy)
    return RPCServer(transport, target, dispatcher, executor)

class RPCServer(msg_server.MessageHandlingServer):
    def __init__(self, transport, target, dispatcher, executor='blocking'):
        super(RPCServer, self).__init__(transport, dispatcher, executor)
        if not isinstance(transport, msg_transport.RPCTransport):
            LOG.warning("Using notification transport for RPC. Please use "
                        "get_rpc_transport to obtain an RPC transport "
                        "instance.")
        self._target = target

RPCServer 继承自 MessageHandlingServer 继承自 ServiceBase,_OrderedTaskRunner

同样的，transport和target是必须要有的，获得 rpc sever实例后，rpc server调用start方法，最终调用到了基类的start方法

@six.add_metaclass(abc.ABCMeta)
class MessageHandlingServer(service.ServiceBase, _OrderedTaskRunner):
    @ordered(reset_after='stop')
    def start(self, override_pool_size=None):
        if self._started:
            LOG.warning('The server has already been started. Ignoring '
                        'the redundant call to start().')
            return

        self._started = True

        executor_opts = {}

        if self.executor_type in ("threading", "eventlet"):
            executor_opts["max_workers"] = (
                override_pool_size or self.conf.executor_thread_pool_size
            )
        self._work_executor = self._executor_cls(**executor_opts)

        try:
            self.listener = self._create_listener()
        except driver_base.TransportDriverError as ex:
            raise ServerListenError(self.target, ex)

        self.listener.start(self._on_incoming)

关键看下 20 行到 25 行，因为这里实例化的是 RPCServer，所以_create_listener 调用的是 RPCServer 的方法

1
2
3

class RPCServer(msg_server.MessageHandlingServer):
    def _create_listener(self):
        return self.transport._listen(self._target, 1, None)

class Transport(object):
    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)
        return self._driver.listen(target, batch_size,
                                   batch_timeout)

class AMQPDriverBase(base.BaseDriver):
    def listen(self, target, batch_size, batch_timeout):
        conn = self._get_connection(rpc_common.PURPOSE_LISTEN)

        listener = RpcAMQPListener(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)

        return base.PollStyleListenerAdapter(listener, batch_size,
                                             batch_timeout)

listen方法实际上关键是执行了三个declare，以下称作：
declare_topic_consumer（1）
declare_topic_consumer (2)
declare_fanout_consumer

declare_topic_consumer（1）

oslo_messaging/_drivers/impl_rabbit.py

class Connection(object):
		def declare_topic_consumer(self, exchange_name, topic, callback=None,
                               queue_name=None):
        """Create a 'topic' consumer."""
        consumer = Consumer(exchange_name=exchange_name,
                            queue_name=queue_name or topic,
                            routing_key=topic,
                            type='topic',
                            durable=self.amqp_durable_queues,
                            exchange_auto_delete=self.amqp_auto_delete,
                            queue_auto_delete=self.amqp_auto_delete,
                            callback=callback,
                            rabbit_ha_queues=self.rabbit_ha_queues)

        self.declare_consumer(consumer)

def declare_consumer(self, consumer):
    """Create a Consumer using the class that was passed in and
    add it to our list of consumers
    """

    def _connect_error(exc):
        log_info = {'topic': consumer.routing_key, 'err_str': exc}
        LOG.error("Failed to declare consumer for topic '%(topic)s': "
                  "%(err_str)s", log_info)

    def _declare_consumer():
        consumer.declare(self)
        tag = self._active_tags.get(consumer.queue_name)
        if tag is None:
            tag = next(self._tags)
            self._active_tags[consumer.queue_name] = tag
            self._new_tags.add(tag)

        self._consumers[consumer] = tag
        return consumer

    with self._connection_lock:
        return self.ensure(_declare_consumer,
                           error_callback=_connect_error)

def declare(self, conn):
    """Re-declare the queue after a rabbit (re)connect."""

    self.queue = kombu.entity.Queue(
        name=self.queue_name,
        channel=conn.channel,
        exchange=self.exchange,
        durable=self.durable,
        auto_delete=self.queue_auto_delete,
        routing_key=self.routing_key,
        queue_arguments=self.queue_arguments)

    try:
        LOG.debug('[%s] Queue.declare: %s',
                  conn.connection_id, self.queue_name)
        self.queue.declare()

@python_2_unicode_compatible
class Queue(MaybeChannelBound):
		def declare(self, nowait=False, channel=None):
        """Declare queue and exchange then binds queue to exchange."""
        if not self.no_declare:
            # - declare main binding.
            self._create_exchange(nowait=nowait, channel=channel)
            self._create_queue(nowait=nowait, channel=channel)
            self._create_bindings(nowait=nowait, channel=channel)
        return self.name

def _create_exchange(self, nowait=False, channel=None):
    if self.exchange:
        self.exchange.declare(nowait=nowait, channel=channel)

def _create_queue(self, nowait=False, channel=None):
    self.queue_declare(nowait=nowait, passive=False, channel=channel)
    if self.exchange and self.exchange.name:
        self.queue_bind(nowait=nowait, channel=channel)

def _create_bindings(self, nowait=False, channel=None):
    for B in self.bindings:
        channel = channel or self.channel
        B.declare(channel)
        B.bind(self, nowait=nowait, channel=channel)

一目了然，总结来说就是使用target的exchange（默认openstack）做为exchange_name
使用target的topic做为默认的queue_name，然后declare这个exchange和queue，然后将二者bind。

declare_topic_consumer（2）
与declare_topic_consumer(1)的唯一区别是这里使用了target.topic结合target.server做为了默认的queue_name。

declare_fanout_consumer

class Connection(object):
		def declare_fanout_consumer(self, topic, callback):
        """Create a 'fanout' consumer."""

        unique = uuid.uuid4().hex
        exchange_name = '%s_fanout' % topic
        queue_name = '%s_fanout_%s' % (topic, unique)

        consumer = Consumer(exchange_name=exchange_name,
                            queue_name=queue_name,
                            routing_key=topic,
                            type='fanout',
                            durable=False,
                            exchange_auto_delete=True,
                            queue_auto_delete=False,
                            callback=callback,
                            rabbit_ha_queues=self.rabbit_ha_queues,
                            rabbit_queue_ttl=self.rabbit_transient_queues_ttl)

        self.declare_consumer(consumer)

与上述两种的区别是，这里的queue_name变成了target.topic+”fanout“+uuid
exchange_name变成了target.topic+”_fanout”，exchange的type变成了fanout

这里也指定了routing_key ，我觉得应该是没用的。

小结

server监听

一、
1.由target.exchange或配置文件（openstack为默认值）命名的exchange（type为topic模式）
2.以target.topic做为queue_name
2.以target.topic做为routing_key进行queue和exchange的绑定
二、
1.由target.exchange或配置文件（openstack为默认值）命名的exchange（type为topic模式）
2.以target.topic结合target.server做为queue_name
2.以target.topic结合target.server做为routingkey进行queue和exchange的绑定
三、
1.由target.topic+”fanout”命名的exchange（type为fanout模式）
2.以target.topic+”fanout“+唯一uuid做为queue_name
2.将queue和exchange的绑定

client调用

1.非阻塞调用：client.cast
2.阻塞调用：client.call

通过prepare来改变client的target进行fanout或者指定server的调用

notification

notification，顾名思义，消息/通知，其概念被分为
通知方:notifier（官方也叫driver）, 监听方:notification_listener
使用时，监听方listener.start, 调用方notifier.notfiy(具体暴漏给用户使用时是sample,audit,info等不同level的方法) 即可把消息发给监听方进行处理。

发送方

以 ceilometer 为例，首先需要实例化一个 Notifier 对象

self.notifier = oslo_messaging.Notifier(
    messaging.get_transport(),
    driver=cfg.CONF.publisher_notifier.telemetry_driver,
    publisher_id="ceilometer.polling")

在处理完数据之后需要将数据发送出去（具体中间处理数据的部分就不讲了，见我的另一篇 polling 源码分析）

1
2
3

class Notifier(object):
    def sample(self, ctxt, event_type, payload):
        self._notify(ctxt, event_type, payload, 'SAMPLE')

def _notify(self, ctxt, event_type, payload, priority, publisher_id=None,
            retry=None):
    payload = self._serializer.serialize_entity(ctxt, payload)
    ctxt = self._serializer.serialize_context(ctxt)

    msg = dict(message_id=six.text_type(uuid.uuid4()),
               publisher_id=publisher_id or self.publisher_id,
               event_type=event_type,
               priority=priority,
               payload=payload,
               timestamp=six.text_type(timeutils.utcnow()))

    def do_notify(ext):
        try:
            ext.obj.notify(ctxt, msg, priority, retry or self.retry)
        except Exception as e:
            _LOG.exception("Problem '%(e)s' attempting to send to "
                           "notification system. Payload=%(payload)s",
                           {'e': e, 'payload': payload})

    if self._driver_mgr.extensions:
        self._driver_mgr.map(do_notify)

class ExtensionManager(object):
    def map(self, func, *args, **kwds):
        if not self.extensions:
            # FIXME: Use a more specific exception class here.
            raise NoMatches('No %s extensions found' % self.namespace)
        response = []
        for e in self.extensions:
            self._invoke_one_plugin(response.append, func, e, args, kwds)
        return response

分析def map：
(Pdb) p func
<function do_notify at 0x7f2c3c1df758>
(Pdb) p args
()
(Pdb) p kwds
{}

2)
(Pdb) p self.extensions
[<stevedore.extension.Extension object at 0x7f2c48771450>]
(Pdb) p e
<stevedore.extension.Extension object at 0x7f2c48771450>
(Pdb) p e.__dict__
{'obj': <oslo_messaging.notify.messaging.MessagingV2Driver object at 0x7f2c48771210>, 'entry_point': EntryPoint.parse('messagingv2 = oslo_messaging.notify.messaging:MessagingV2Driver'), 'name': 'messagingv2', 'plugin': <class 'oslo_messaging.notify.messaging.MessagingV2Driver'>}

class ExtensionManager(object):
		def _invoke_one_plugin(self, response_callback, func, e, args, kwds):
        try:
            response_callback(func(e, *args, **kwds))
        except Exception as err:
            if self.propagate_map_exceptions:
                raise
            else:
                LOG.error('error calling %r: %s', e.name, err)
                LOG.exception(err)

可以看到 response_callback 传过来是一个 append 方法，这里的 func 就是 do_notify，发生调用，即

def _notify(self, ctxt, event_type, payload, priority, publisher_id=None,
            retry=None):
    payload = self._serializer.serialize_entity(ctxt, payload)
    ctxt = self._serializer.serialize_context(ctxt)

    msg = dict(message_id=six.text_type(uuid.uuid4()),
               publisher_id=publisher_id or self.publisher_id,
               event_type=event_type,
               priority=priority,
               payload=payload,
               timestamp=six.text_type(timeutils.utcnow()))

    def do_notify(ext):
        try:
            ext.obj.notify(ctxt, msg, priority, retry or self.retry)
        except Exception as e:
            _LOG.exception("Problem '%(e)s' attempting to send to "
                           "notification system. Payload=%(payload)s",
                           {'e': e, 'payload': payload})

    if self._driver_mgr.extensions:
        self._driver_mgr.map(do_notify)

即调用 13 行方法

(Pdb) p ext
<stevedore.extension.Extension object at 0x7f2c48771450>
(Pdb) p ext.__dict__
{'obj': <oslo_messaging.notify.messaging.MessagingV2Driver object at 0x7f2c48771210>, 'entry_point': EntryPoint.parse('messagingv2 = oslo_messaging.notify.messaging:MessagingV2Driver'), 'name': 'messagingv2', 'plugin': <class 'oslo_messaging.notify.messaging.MessagingV2Driver'>}

(Pdb) p ext.obj
<oslo_messaging.notify.messaging.MessagingV2Driver object at 0x7f2c48771210>
(Pdb) p ext.obj.__dict__
{'topics': ['notifications'], 'version': 2.0, 'transport': <oslo_messaging.transport.Transport object at 0x7f2c485df890>, 'conf': <oslo_config.cfg.ConfigOpts object at 0x16ba490>}

class MessagingDriver(notifier.Driver):
    def __init__(self, conf, topics, transport, version=1.0):
        super(MessagingDriver, self).__init__(conf, topics, transport)
        self.version = version

    def notify(self, ctxt, message, priority, retry):
        priority = priority.lower()
        for topic in self.topics:
            target = oslo_messaging.Target(topic='%s.%s' % (topic, priority))
            try:
                self.transport._send_notification(target, ctxt, message,
                                                  version=self.version,
                                                  retry=retry)
            except Exception:
                LOG.exception("Could not send notification to %(topic)s. "
                              "Payload=%(message)s",
                              {'topic': topic, 'message': message})


class MessagingV2Driver(MessagingDriver):

    "Send notifications using the 2.0 message format."

    def __init__(self, conf, **kwargs):
        super(MessagingV2Driver, self).__init__(conf, version=2.0, **kwargs)

实际调用到第 6 行，最关键的地方:
43 -> priority = priority.lower()
44 for topic in self.topics:
45 target = oslo_messaging.Target(topic=’%s.%s’ % (topic, priority))

可以看到oslo_messaging封装得到的真正队列名称是:

. 这种形式，
样例:
notifications.sample
所以，一旦调用oslo_messaging.notifier.sample来发送消息，此时的priority就被设置为
sample了，并且该优先级被用于最终拼接生成oslo_messaging的Target，最后生成了对应的
队列notifications.sample

class Transport(object):
    def _send_notification(self, target, ctxt, message, version, retry=None):
        if not target.topic:
            raise exceptions.InvalidTarget('A topic is required to send',
                                           target)
        self._driver.send_notification(target, ctxt, message, version,
                                       retry=retry)

(Pdb) p self._driver
<oslo_messaging._drivers.impl_rabbit.RabbitDriver object at 0x7f2c4876de90>
(Pdb) p self._driver.__dict__
{'_waiter': None, '_allowed_remote_exmods': [], '_reply_q_lock': <thread.lock object at 0x7f2c482efc30>, 'conf': <oslo_config.cfg.ConfigOpts object at 0x16ba490>, '_default_exchange': 'ceilometer', '_connection_pool': <oslo_messaging._drivers.pool.ConnectionPool object at 0x7f2c4876df90>, '_reply_q': None, 'missing_destination_retry_timeout': 60, 'prefetch_size': 0, '_reply_q_conn': None, '_url': <TransportURL transport='rabbit', hosts=[<TransportHost hostname='rabbitmq.openstack.svc.cluster.local', port=5672, username='rabbitmq', password='vut8mvvS'>]>}

class AMQPDriverBase(base.BaseDriver):
    def send_notification(self, target, ctxt, message, version, retry=None):
        return self._send(target, ctxt, message,
                          envelope=(version == 2.0), notify=True, retry=retry)

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

    msg = message

    if wait_for_reply:
        msg_id = uuid.uuid4().hex
        msg.update({'_msg_id': msg_id})
        msg.update({'_reply_q': self._get_reply_q()})
        msg.update({'_timeout': call_monitor_timeout})

    rpc_amqp._add_unique_id(msg)
    unique_id = msg[rpc_amqp.UNIQUE_ID]

    rpc_amqp.pack_context(msg, ctxt)

    if envelope:
        msg = rpc_common.serialize_msg(msg)

    if wait_for_reply:
        self._waiter.listen(msg_id)
        log_msg = "CALL msg_id: %s " % msg_id
    else:
        log_msg = "CAST unique_id: %s " % unique_id

    try:
        with self._get_connection(rpc_common.PURPOSE_SEND) as conn:
            if notify:
                exchange = self._get_exchange(target)
                LOG.debug(log_msg + "NOTIFY exchange '%(exchange)s'"
                          " topic '%(topic)s'", {'exchange': exchange,
                                                 'topic': target.topic})
                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.debug(log_msg + "exchange '%(exchange)s'"
                          " topic '%(topic)s'", {'exchange': exchange,
                                                 'topic': topic})
                conn.topic_send(exchange_name=exchange, topic=topic,
                                msg=msg, timeout=timeout, retry=retry,
                                transport_options=transport_options)

        if wait_for_reply:
            result = self._waiter.wait(msg_id, timeout,
                                       call_monitor_timeout)
            if isinstance(result, Exception):
                raise result
            return result
    finally:
        if wait_for_reply:
            self._waiter.unlisten(msg_id)

到了这里就和 rpc client 那块差不多了，区别就是这里发送的时候，传了 notify=True，后面就不啰嗦了，和前面差不多

接收/监听方

还是以 ceilometer 为例

urls = cfg.CONF.notification.messaging_urls or [None]
for url in urls:
    transport = messaging.get_transport(url)
    # NOTE(gordc): ignore batching as we want pull
    # to maintain sequencing as much as possible.
    listener = messaging.get_batch_notification_listener(
        transport, targets, endpoints)
    listener.start()
    self.listeners.append(listener)

(Pdb) endpoints
[<ceilometer.event.endpoint.EventsNotificationEndpoint object at 0x2f5f610>,
 <ceilometer.ipmi.notifications.ironic.TemperatureSensorNotification object at 0x7fa1741f6810>,
  <ceilometer.telemetry.notifications.TelemetryIpc object at 0x7fa17424d410>,
   <ceilometer.ipmi.notifications.ironic.FanSensorNotification object at 0x7fa17424d8d0>,
    <ceilometer.ipmi.notifications.ironic.VoltageSensorNotification object at 0x7fa1742461d0>,
     <ceilometer.meter.notifications.ProcessMeterNotifications object at 0x7fa17424dc90>,
      <ceilometer.ipmi.notifications.ironic.CurrentSensorNotification object at 0x7fa17467ddd0>]
(Pdb) cfg.CONF.notification.messaging_urls
['rabbit://rabbitmq:I5dZs2KN@rabbitmq.openstack.svc.cluster.local:5672/']

def get_batch_notification_listener(transport, targets, endpoints,
                                    allow_requeue=False,
                                    batch_size=1, batch_timeout=None):
    return oslo_messaging.get_batch_notification_listener(
        transport, targets, endpoints, executor='threading',
        allow_requeue=allow_requeue,
        batch_size=batch_size, batch_timeout=batch_timeout)

def get_batch_notification_listener(transport, targets, endpoints,
                                    executor='blocking', serializer=None,
                                    allow_requeue=False, pool=None,
                                    batch_size=None, batch_timeout=None):
    dispatcher = notify_dispatcher.BatchNotificationDispatcher(
        endpoints, serializer)
    return BatchNotificationServer(
        transport, targets, dispatcher, executor, allow_requeue, pool,
        batch_size, batch_timeout
    )

其中 BatchNotificationDispatcher 实现了 dispatch 方法，BatchNotificationServer 中实现了 _process_incoming方法

BatchNotificationDispatcher 继承自 NotificationDispatcher 继承自 DispatcherBase

BatchNotificationServer 继承自 NotificationServerBase 继承自 MessageHandlingServer 继承自 ServiceBase，_OrderedTaskRunner

因为 listen 实例化的是 BatchNotificationServer 对象，所以在调用 listener.start() 的时候，实际调用的是 MessageHandlingServer 的 start 方法

@six.add_metaclass(abc.ABCMeta)
class MessageHandlingServer(service.ServiceBase, _OrderedTaskRunner):
		@ordered(reset_after='stop')
    def start(self, override_pool_size=None):
        if self._started:
            LOG.warning('The server has already been started. Ignoring '
                        'the redundant call to start().')
            return

        self._started = True

        executor_opts = {}

        if self.executor_type in ("threading", "eventlet"):
            executor_opts["max_workers"] = (
                override_pool_size or self.conf.executor_thread_pool_size
            )
        self._work_executor = self._executor_cls(**executor_opts)

        try:
            self.listener = self._create_listener()
        except driver_base.TransportDriverError as ex:
            raise ServerListenError(self.target, ex)

        self.listener.start(self._on_incoming)

重点在 21 和 25 行，第 21 行调到了

class NotificationServerBase(msg_server.MessageHandlingServer):
    def __init__(self, transport, targets, dispatcher, executor='blocking',
                 allow_requeue=True, pool=None, batch_size=1,
                 batch_timeout=None):
        super(NotificationServerBase, self).__init__(transport, dispatcher,
                                                     executor)
        self._allow_requeue = allow_requeue
        self._pool = pool
        self.targets = targets
        self._targets_priorities = set(
            itertools.product(self.targets,
                              self.dispatcher.supported_priorities)
        )

        self._batch_size = batch_size
        self._batch_timeout = batch_timeout

    def _create_listener(self):
        return self.transport._listen_for_notifications(
            self._targets_priorities, self._pool, self._batch_size,
            self._batch_timeout
        )

def _listen_for_notifications(self, targets_and_priorities, pool,
                              batch_size, batch_timeout):
    for target, priority in targets_and_priorities:
        if not target.topic:
            raise exceptions.InvalidTarget('A target must have '
                                           'topic specified',
                                           target)
    return self._driver.listen_for_notifications(
        targets_and_priorities, pool, batch_size, batch_timeout
    )

class AMQPDriverBase(base.BaseDriver):
		def listen_for_notifications(self, targets_and_priorities, pool,
                                 batch_size, batch_timeout):
        conn = self._get_connection(rpc_common.PURPOSE_LISTEN)

        listener = NotificationAMQPListener(self, conn)
        for target, priority in targets_and_priorities:
            conn.declare_topic_consumer(
                exchange_name=self._get_exchange(target),
                topic='%s.%s' % (target.topic, priority),
                callback=listener, queue_name=pool)
        return base.PollStyleListenerAdapter(listener, batch_size,
                                             batch_timeout)

在listen_for_notification中建立连接后，创建topic.priority的queue。PollStyleListenerAdapter启动一个线程对获取到的数据进行处理，此处返回此类，所以_create_listener 就是返回了一个PollStyleListenerAdapter的实例

然后调用 self.listener.start(self._on_incoming)

PollStyleListenerAdapter 继承自 Listener

class PollStyleListenerAdapter(Listener):
    """A Listener that uses a PollStyleListener for message transfer. A
    dedicated thread is created to do message polling.
    """

    def __init__(self, poll_style_listener, batch_size, batch_timeout):
        super(PollStyleListenerAdapter, self).__init__(
            batch_size, batch_timeout, poll_style_listener.prefetch_size
        )
        self._poll_style_listener = poll_style_listener
        self._listen_thread = threading.Thread(target=self._runner)
        self._listen_thread.daemon = True
        self._started = False

    def start(self, on_incoming_callback):
        super(PollStyleListenerAdapter, self).start(on_incoming_callback)
        self._started = True
        self._listen_thread.start()

    @excutils.forever_retry_uncaught_exceptions
    def _runner(self):
        while self._started:
            incoming = self._poll_style_listener.poll(
                batch_size=self.batch_size, batch_timeout=self.batch_timeout)

            if incoming:
                self.on_incoming_callback(incoming)

        # listener is stopped but we need to process all already consumed
        # messages
        while True:
            incoming = self._poll_style_listener.poll(
                batch_size=self.batch_size, batch_timeout=self.batch_timeout)

            if not incoming:
                return
            self.on_incoming_callback(incoming)

    def stop(self):
        self._started = False
        self._poll_style_listener.stop()
        self._listen_thread.join()
        super(PollStyleListenerAdapter, self).stop()

    def cleanup(self):
        self._poll_style_listener.cleanup()

NotificationAMQPListener 继承自 AMQPListener 继承自 PollStyleListener

可以看到这里的 start 方法就是调用了 _runner方法，这里的 self.poll_style_listener 就是 NotificationAMQPListener 实例，通过 poll 不断从队列中取出数据

即调用 AMQPListener.poll

class NotificationAMQPListener(AMQPListener):
    message_cls = NotificationAMQPIncomingMessage
    
class AMQPListener(base.PollStyleListener):

    def __init__(self, driver, conn):
        super(AMQPListener, self).__init__(driver.prefetch_size)
        self.driver = driver
        self.conn = conn
        self.msg_id_cache = rpc_amqp._MsgIdCache()
        self.incoming = []
        self._shutdown = threading.Event()
        self._shutoff = threading.Event()
        self._obsolete_reply_queues = ObsoleteReplyQueuesCache()
        self._message_operations_handler = MessageOperationsHandler(
            "AMQPListener")
        self._current_timeout = ACK_REQUEUE_EVERY_SECONDS_MIN

    def __call__(self, message):
        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(self.message_cls(
            self,
            ctxt.to_dict(),
            message,
            unique_id,
            ctxt.msg_id,
            ctxt.reply_q,
            ctxt.client_timeout,
            self._obsolete_reply_queues,
            self._message_operations_handler))

    @base.batch_poll_helper
    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:
                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:
                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()

在 call 方法又调用了 self.message_cls ，即 NotificationAMQPIncomingMessage

class NotificationAMQPIncomingMessage(AMQPIncomingMessage):
    def acknowledge(self):
        def _do_ack():
            try:
                self.message.acknowledge()
            except Exception as exc:
                # NOTE(kgiusti): this failure is likely due to a loss of the
                # connection to the broker.  Not much we can do in this case,
                # especially considering the Notification has already been
                # dispatched. This *could* result in message duplication
                # (unacked msg is returned to the queue by the broker), but the
                # driver tries to catch that using the msg_id_cache.
                LOG.warning("Failed to acknowledge received message: %s", exc)
        self._message_operations_handler.do(_do_ack)
        self.listener.msg_id_cache.add(self.unique_id)

    def requeue(self):
        # NOTE(sileht): In case of the connection is lost between receiving the
        # message and requeing it, this requeue call fail
        # but because the message is not acknowledged and not added to the
        # msg_id_cache, the message will be reconsumed, the only difference is
        # the message stay at the beginning of the queue instead of moving to
        # the end.
        def _do_requeue():
            try:
                self.message.requeue()
            except Exception as exc:
                LOG.warning("Failed to requeue received message: %s", exc)
        self._message_operations_handler.do(_do_requeue)

NotificationAMQPIncomingMessage 继承自 AMQPIncomingMessage 继承自 RpcIncomingMessage 继承自 IncomingMessage

（self.conn.consume 中有回调函数调用，call 方法，往 incoming 中塞值，poll 再从里面取值处理）括号中的部分有点疑问，理得不是很清楚，如果有清楚的大佬，欢迎指教。

NotificationAMQPIncomingMessage 是一个消息对象，包含了acknowledge和requeue方法，可以用于消息确认或再次入队。

取出之后用self.on_incoming_callback(incoming)处理，即MessageHandlingServer中的self.__on_incoming处理

@six.add_metaclass(abc.ABCMeta)
class MessageHandlingServer(service.ServiceBase, _OrderedTaskRunner):
		def _on_incoming(self, incoming):
        """Handles on_incoming event

        :param incoming: incoming request.
        """
        self._work_executor.submit(self._process_incoming, incoming)

我们看到这边实际使用的是self._process_incoming来处理，即采用BatchNotificationServer类中的self.__process_incoming处理

class BatchNotificationServer(NotificationServerBase):

    def _process_incoming(self, incoming):
        try:
            not_processed_messages = self.dispatcher.dispatch(incoming)
        except Exception:
						......