Как изящно приостановить и возобновить потребление в rabbitmq, pika python

Question

Я использую basic_consume() для получения сообщений и basic_cancel для отмены потребления, но есть проблема.

Вот код pika.channel

 def basic_consume(self, consumer_callback, queue='', no_ack=False,
                      exclusive=False, consumer_tag=None):
        """Sends the AMQP command Basic.Consume to the broker and binds messages
        for the consumer_tag to the consumer callback. If you do not pass in
        a consumer_tag, one will be automatically generated for you. Returns
        the consumer tag.

        For more information on basic_consume, see:
        http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.consume

        :param method consumer_callback: The method to callback when consuming
        :param queue: The queue to consume from
        :type queue: str or unicode
        :param bool no_ack: Tell the broker to not expect a response
        :param bool exclusive: Don't allow other consumers on the queue
        :param consumer_tag: Specify your own consumer tag
        :type consumer_tag: str or unicode
        :rtype: str

        """
        self._validate_channel_and_callback(consumer_callback)

        # If a consumer tag was not passed, create one
        consumer_tag = consumer_tag or 'ctag%i.%s' % (self.channel_number,
                                                      uuid.uuid4().get_hex())

        if consumer_tag in self._consumers or consumer_tag in self._cancelled:
            raise exceptions.DuplicateConsumerTag(consumer_tag)

        self._consumers[consumer_tag] = consumer_callback
        self._pending[consumer_tag] = list()
        self._rpc(spec.Basic.Consume(queue=queue,
                                     consumer_tag=consumer_tag,
                                     no_ack=no_ack,
                                     exclusive=exclusive),
                           self._on_eventok,
                           [(spec.Basic.ConsumeOk,
                             {'consumer_tag': consumer_tag})])

        return consumer_tag

def basic_cancel(self, callback=None, consumer_tag='', nowait=False):
        """This method cancels a consumer. This does not affect already
        delivered messages, but it does mean the server will not send any more
        messages for that consumer. The client may receive an arbitrary number
        of messages in between sending the cancel method and receiving the
        cancel-ok reply. It may also be sent from the server to the client in
        the event of the consumer being unexpectedly cancelled (i.e. cancelled
        for any reason other than the server receiving the corresponding
        basic.cancel from the client). This allows clients to be notified of
        the loss of consumers due to events such as queue deletion.

        :param method callback: Method to call for a Basic.CancelOk response
        :param str consumer_tag: Identifier for the consumer
        :param bool nowait: Do not expect a Basic.CancelOk response
        :raises: ValueError

        """
        self._validate_channel_and_callback(callback)
        if consumer_tag not in self.consumer_tags:
            return
        if callback:
            if nowait is True:
                raise ValueError('Can not pass a callback if nowait is True')
            self.callbacks.add(self.channel_number,
                               spec.Basic.CancelOk,
                               callback)
        self._cancelled.append(consumer_tag)
        self._rpc(spec.Basic.Cancel(consumer_tag=consumer_tag,
                                    nowait=nowait),
                  self._on_cancelok,
                  [(spec.Basic.CancelOk,
                    {'consumer_tag': consumer_tag})] if nowait is False else [])

Как вы можете видеть каждый раз, когда я отменяю потребление, потребительский тег добавляется в список _canceled. И если бы я снова использовал этот тег в basic_consume, то возникнет исключение duplicateConsumer. Ну, я мог бы каждый раз использовать новый тег customer_tag, но на самом деле это не так. Потому что рано или поздно сгенерированный тег будет точно соответствовать некоторым из предыдущих.

Как я должен приостановить и возобновить потребление изящно в Пика?

Answer 1

Похоже, что Pika делает больше, чем следовало бы - ей не нужно создавать потребительский тег, если он не предоставлен (сервер будет), и также не нужно отслеживать дублированные потребительские теги (возобновляя с тем же тегом поддерживается сервером).

Так что я не уверен, как это сделать с Пика - я полагаю, ошибка.

Answer 2

Есть ли причина, по которой вы определяете свою собственную consumer_tags? Вы можете передать пустую строку и позволить RabbitMQ сгенерировать потребительские теги для вас. Ответ от basic.consume, который basic.consume-ok вернет сгенерированный consumer_tag, так что вы можете использовать его позже, чтобы перестать потреблять.

Смотрите: http://www.rabbitmq.com/amqp-0-9-1-reference.html