#: This is the callback applied when a message is received. def handle_message(body, message): print(f'Received message: {body!r}') print(' properties:\n{}'.format(pretty(message.properties))) print(' delivery_info:\n{}'.format(pretty(message.delivery_info))) message.ack()
with Connection('amqp://guest:guest@localhost:5672//') as connection:
with Consumer(connection, queue, callbacks=[handle_message]):
def __init__(self, channel, queues=None, no_ack=None, auto_declare=None, callbacks=None, on_decode_error=None, on_message=None, accept=None, prefetch_count=None, tag_prefix=None): self.channel = channel # Queue的列表 self.queues = maybe_list(queues or [])
self.no_ack = self.no_ack if no_ack is None else no_ack # 消息的回调函数 self.callbacks = (self.callbacks or [] if callbacks is None else callbacks) # 自定义的消息处理方法 self.on_message = on_message self.tag_prefix = tag_prefix self._active_tags = {} ...
if self.channel: self.revive(self.channel)
def revive(self, channel): """Revive consumer after connection loss.""" self._active_tags.clear() channel = self.channel = maybe_channel(channel) # modify dict size while iterating over it is not allowed for qname, queue in list(self._queues.items()): # name may have changed after declare self._queues.pop(qname, None) queue = self._queues[queue.name] = queue(self.channel) # queue和channel绑定 queue.revive(channel) ...
def consume(self, no_ack=None): tag = self._add_tag(queue, consumer_tag) # 每个queue消息消息 for queue in self._queues: queue.consume(tag, self._receive_callback, no_ack=no_ack, nowait=nowait)
def _receive_callback(self, message): accept = self.accept on_m, channel, decoded = self.on_message, self.channel, None try: ... # 消息反序列化 decoded = None if on_m else message.decode() except Exception as exc: if not self.on_decode_error: raise self.on_decode_error(message, exc) else: return on_m(message) if on_m else self.receive(decoded, message)
def receive(self, body, message): """Method called when a message is received.
This dispatches to the registered :attr:`callbacks`.
Arguments: body (Any): The decoded message body. message (~kombu.Message): The message instance.
Raises: NotImplementedError: If no consumer callbacks have been registered. """ # 执行callback callbacks = self.callbacks ... # 默认就是body和message回传给业务函数 [callback(body, message) for callback in callbacks]
def maybe_bind(self, channel): """Bind instance to channel if not already bound.""" if not self.is_bound and channel: self._channel = maybe_channel(channel) self.when_bound() self._is_bound = True return self
@property def is_bound(self): """Flag set if the channel is bound.""" return self._is_bound and self._channel is not None
exchange对象的创建和绑定到channel:
class Exchange(MaybeChannelBound): def __init__(self, name='', type='', channel=None, **kwargs): super().__init__(**kwargs) self.name = name or self.name self.type = type or self.type self.maybe_bind(channel) ...
def declare(self, nowait=False, passive=None, channel=None): """Declare the exchange.
Creates the exchange on the broker, unless passive is set in which case it will only assert that the exchange exists.
Argument: nowait (bool): If set the server will not respond, and a response will not be waited for. Default is :const:`False`. """ if self._can_declare(): passive = self.passive if passive is None else passive # 依托于channel return (channel or self.channel).exchange_declare( exchange=self.name, type=self.type, durable=self.durable, auto_delete=self.auto_delete, arguments=self.arguments, nowait=nowait, passive=passive, )
queue对象创建完成后也需要绑定到channel:
class Queue(MaybeChannelBound): def __init__(self, name='', exchange=None, routing_key=''
, channel=None, bindings=None, on_declared=None, **kwargs): super().__init__(**kwargs) self.name = name or self.name self.maybe_bind(channel) ...
然后申明queue，这个过程包括下面3个步骤:
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: # 隐式申明exchange self.exchange.declare(nowait=nowait, channel=channel)
def _create_queue(self, nowait=False, channel=None): # 申明queue self.queue_declare(nowait=nowait, passive=False, channel=channel) if self.exchange and self.exchange.name: # 绑定queue和exchange 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)
def _init_params(self, hostname, userid, password, virtual_host, port, insist, ssl, transport, connect_timeout, login_method, heartbeat): transport = transport or 'amqp' if transport == 'amqp' and supports_librabbitmq(): transport = 'librabbitmq' if transport == 'rediss' and ssl_available and not ssl: logger.warning( 'Secure redis scheme specified (rediss) with no ssl ' 'options, defaulting to insecure SSL behaviour.' ) ssl = {'ssl_cert_reqs': CERT_NONE} self.hostname = hostname self.userid = userid self.password = password self.login_method = login_method # 虚拟主机隔离 self.virtual_host = virtual_host or self.virtual_host self.port = port or self.port self.insist = insist self.connect_timeout = connect_timeout self.ssl = ssl # 传输类 self.transport_cls = transport self.heartbeat = heartbeat and float(heartbeat)
Warning: This instance is transport specific, so do not depend on the interface of this object. """ if not self._closed: if not self.connected: # 创建连接 return self._ensure_connection( max_retries=1, reraise_as_library_errors=False ) return self._connection
Created upon access and closed when the connection is closed.
Note: Can be used for automatic channel handling when you only need one channel, and also it is the channel implicitly used if a connection is passed instead of a channel, to functions that require a channel. """ # make sure we're still connected, and if not refresh. conn_opts = self._extract_failover_opts() # 创建连接 self._ensure_connection(**conn_opts)
if self._default_channel is None: self._default_channel = self.channel() return self._default_channel
连接创建完成后，继续创建channel:
def channel(self): """Create and return a new channel.""" self._debug('create channel') chan = self.transport.create_channel(self.connection) return chan
def get_transport_cls(self): """Get the currently used transport class.""" transport_cls = self.transport_cls if not transport_cls or isinstance(transport_cls, str): transport_cls = get_transport_cls(transport_cls) return transport_cls
The fastest serialization method, but restricts you to python clients. """ def pickle_dumps(obj, dumper=pickle.dumps): return dumper(obj, protocol=pickle_protocol)
* integers, floating point numbers, complex numbers
* strings, bytes, bytearrays
* tuples, lists, sets, and dictionaries containing only picklable objects
* functions defined at the top level of a module (using def, not lambda)
* built-in functions defined at the top level of a module
* classes that are defined at the top level of a module
* instances of such classes whose __dict__ or the result of calling __getstate__() is picklable (see section Pickling Class Instances for details).
配置类的简化
Object提供了一种快速构建对象的方法:
class Object: """Common base class.
Supports automatic kwargs->attributes handling, and cloning. """
attrs = ()
def __init__(self, *args, **kwargs): # attrs 在子类中定义 for name, type_ in self.attrs: value = kwargs.get(name) # 从字典参数给属性动态赋值 if value is not None: setattr(self, name, (type_ or _any)(value)) else: try: getattr(self, name) except AttributeError: setattr(self, name, None)
Queue展示了这种方式的示例，比如max_length属性:
class Queue(MaybeChannelBound): attrs = ( .. ('max_length', int), ... ) def __init__(self, name='', exchange=None, routing_key='', channel=None, bindings=None, on_declared=None, **kwargs):
self.name = name or self.name ...
