client.py 文件源码

def run(self):
        self._is_running = True
        while self._is_running:
            if self.consumer.is_connected():
                producer = kombu.Producer(self.consumer._channel, on_return=self.consumer._handle_return)
                try:
                    queued_request = self._out_queue.get(timeout=0.5)
                    if True:
                        # with kombu.producers[self.consumer.get_connection()].acquire(block=True) as producer:
                        # producer.on_return = print
                        try:
                            self._dispatch_request(queued_request, producer)
                        except Exception as e:
                            # except ConnectionResetError:
                            log.debug('Failed to dispatch request, re-enqueueing again, error was: {}'.format(
                                str(e)
                            ))
                            self.enqueue(queued_request)
                except Empty:
                    continue
            else:
                sleep(0.5)
                log.debug('Waiting for consumer to be ready...')