先看一下下面这段代码

  1. from celery import Celery
  4. broker_url = 'amqp://guest:guest@127.0.0.1:5672//'
  8. def get_celery_client(broker_url=broker_url,
  9.                       broker_ssl_enabled=False,
  10.                       broker_ssl_cert_path=None,
  11.                       max_retries=None):
  13.     from celery import Celery
  15.     celery_client = Celery(broker=broker_url,
  16.                            backend=broker_url)
  17.     celery_client.conf.update(
  18.         CELERY_TASK_RESULT_EXPIRES=600)
  20.     if broker_ssl_enabled:
  21.         import ssl
  22.         celery_client.conf.BROKER_USE_SSL = {
  23.             'ca_certs': broker_ssl_cert_path,
  24.             'cert_reqs': ssl.CERT_REQUIRED,
  25.         }
  27.     celery_client.pool.connection.ensure_connection(max_retries=max_retries)
  28.     return celery_client
  32. def ping(worker=None):
  33.     app = get_celery_client(broker_url, max_retries=3)
  34.     try:
  35.         destination = [worker] if worker else []
  36.         inspect = app.control.inspect(destination=[worker],
  37.                                       timeout=10)
  38.         resp = inspect.ping()
  39.     except BaseException:
  40.         return None
  41.     finally:
  42.         if app:
  43.             app.close()
  44.     return resp
  47. resp = ping('celery@test_queue')
  48. print(resp)

乍一看代码36行app.control.inspect(destination=[worker])已经指定了要检查的目标队列
实际上，inspect.ping()方法仍然会将ping消息发送给所有bind celery.pidbox 这个exchange的queue
以下从两个方向分析为什么ping消息是广播的：

  1. 查看RabbitMQ的exchange列表，找到celery.pidbox这个exchange，进到详情页面，可以看到celery.pidbox这个exchange 类型为fanout（广播类型）

image.png

  1. 查看celery源码
  • celery.app.control:Inspect.ping() ```python class Inspect(object): “””API for app.control.inspect.”””

…省略无关代码…

  1. def _request(self, command, **kwargs):
  2.     return self._prepare(self.app.control.broadcast(
  3.         command,
  4.         arguments=kwargs,
  5.         destination=self.destination,
  6.         callback=self.callback,
  7.         connection=self.connection,
  8.         limit=self.limit,
  9.         timeout=self.timeout, reply=True,
  10.         pattern=self.pattern, matcher=self.matcher,
  11.     ))

…省略无关代码…

  1. def ping(self, destination=None):
  2.     return self._request('ping')
  1. ping方法调用了app.control.broadcast方法<br />看到这里你还是会有些诧异，“broadcast方法里面不是可以指定destination嘛，为什么还是广播的呢？”，那么按图索骥，看看这个broadcast到底怎么工作的。
  3. - `celery.app.control:Control.broadcast()`
  4. ```python
  5. class Control(object):
  6.     """Worker remote control client."""
  7. # ...省略无关代码...    
  8.     def broadcast(self, command, arguments=None, destination=None,
  9.                   connection=None, reply=False, timeout=1.0, limit=None,
  10.                   callback=None, channel=None, pattern=None, matcher=None,
  11.                   **extra_kwargs):
  12.         """Broadcast a control command to the celery workers.
  14.         Arguments:
  15.             command (str): Name of command to send.
  16.             arguments (Dict): Keyword arguments for the command.
  17.             destination (List): If set, a list of the hosts to send the
  18.                 command to, when empty broadcast to all workers.
  19.             connection (kombu.Connection): Custom broker connection to use,
  20.                 if not set, a connection will be acquired from the pool.
  21.             reply (bool): Wait for and return the reply.
  22.             timeout (float): Timeout in seconds to wait for the reply.
  23.             limit (int): Limit number of replies.
  24.             callback (Callable): Callback called immediately for
  25.                 each reply received.
  26.             pattern (str): Custom pattern string to match
  27.             matcher (Callable): Custom matcher to run the pattern to match
  28.         """
  29.         with self.app.connection_or_acquire(connection) as conn:
  30.             arguments = dict(arguments or {}, **extra_kwargs)
  31.             if pattern and matcher:
  32.                 # tests pass easier without requiring pattern/matcher to
  33.                 # always be sent in
  34.                 return self.mailbox(conn)._broadcast(
  35.                     command, arguments, destination, reply, timeout,
  36.                     limit, callback, channel=channel,
  37.                     pattern=pattern, matcher=matcher,
  38.                 )
  39.             else:
  40.                 return self.mailbox(conn)._broadcast(
  41.                     command, arguments, destination, reply, timeout,
  42.                     limit, callback, channel=channel,
  43.                 )

我们可以继续查看mailbox._broadcast

  • kombu.pidbox:Mailbox._broadcast

    1. class Mailbox(object)
    2.   def _publish(self, type, arguments, destination=None,
    3.                reply_ticket=None, channel=None, timeout=None,
    4.                serializer=None, producer=None):
    5.       message = {'method': type,
    6.                  'arguments': arguments,
    7.                  'destination': destination}
    8.       chan = channel or self.connection.default_channel
    9.       exchange = self.exchange
    10.       if reply_ticket:
    11.           maybe_declare(self.reply_queue(channel))
    12.           message.update(ticket=reply_ticket,
    13.                          reply_to={'exchange': self.reply_exchange.name,
    14.                                    'routing_key': self.oid})
    15.       serializer = serializer or self.serializer
    16.       with self.producer_or_acquire(producer, chan) as producer:
    17.           producer.publish(
    18.               message, exchange=exchange.name, declare=[exchange],
    19.               headers={'clock': self.clock.forward(),
    20.                        'expires': time() + timeout if timeout else 0},
    21.               serializer=serializer,
    22.           )
    24.   def _broadcast(self, command, arguments=None, destination=None,
    25.                  reply=False, timeout=1, limit=None,
    26.                  callback=None, channel=None, serializer=None):
    27.       if destination is not None and \
    28.               not isinstance(destination, (list, tuple)):
    29.           raise ValueError(
    30.               'destination must be a list/tuple not {0}'.format(
    31.                   type(destination)))
    33.       arguments = arguments or {}
    34.       reply_ticket = reply and uuid() or None
    35.       chan = channel or self.connection.default_channel
    37.       # Set reply limit to number of destinations (if specified)
    38.       if limit is None and destination:
    39.           limit = destination and len(destination) or None
    41.       serializer = serializer or self.serializer
    42.       self._publish(command, arguments, destination=destination,
    43.                     reply_ticket=reply_ticket,
    44.                     channel=chan,
    45.                     timeout=timeout,
    46.                     serializer=serializer)
    48.       if reply_ticket:
    49.           return self._collect(reply_ticket, limit=limit,
    50.                                timeout=timeout,
    51.                                callback=callback,
    52.                                channel=chan)

    从这个_broadcast方法可以看出来，destination参数传给了publish方法，publish方法将destination放入了message body中，并且在调用producer的publish方法时未指定任何routing_key，消息直接发送给了celery.pidbox这个exchange。
    从代码层面分析得知，inspect.ping()方法是将ping消息广播给了所有绑定在celery.pidbox这个exchange上的queue。