上一篇我们介绍了 RabbitMq 的相关知识
今天介绍一下 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:[email protected]:port[,hostN:portN]/virtual_host
目前支持的Transport有rabbit,qpid与zmq,分别对应不同的后端消息总线.用户可以使用oslo.messaging.get_transport函数来获得transport对象实例的句柄.
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'})
发送通知消息时,首先要构造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 调用
@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方法得来的
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 的方法
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做为routing_key进行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 源码分析）
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:[email protected]: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:
......
可以看到该处理函数会调用dispatcher对象来分派消息，这里的self.dispatcher，就是之前使用NotificationDispatcher初始化后传过来的参数，后面就是一些数据处理的过程，详细请见我的另一篇 nogtification 源码分析
参考：
https://www.cnblogs.com/gange111/p/9560446.html
https://blog.csdn.net/MrYuanRs/article/details/105955720?utm_medium=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.channel_param&depth_1-utm_source=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.channel_param
https://blog.csdn.net/qingyuanluofeng/article/details/102511492?utm_medium=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param&depth_1-utm_source=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-2.channel_param
![TestLink导出用例转换工具(XML2Excel)[图]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)