Python:在具有干净调用堆栈的except块上重新运行函数

Question

我有生成潜在无限调用堆栈的代码（简化）：

def listen(self, pipeline):
    try:
        for message in self.channel.consume(self.queue_name):
            pipeline.process(message)
            self.channel.basic_ack(delivery_tag=method_frame.delivery_tag)

    except (pika.exceptions.StreamLostError,
            pika.exceptions.ConnectionClosed,
            pika.exceptions.ChannelClosed,
            ConnectionResetError) as e:
        logging.warning(f'Connection dropped for queue {self.queue_name}. Exception: {e}. Reconnecting...')
        self._reconnect()
        self.listen(pipeline)

如果存在任何网络问题，它将记录错误，重新连接并进一步移动。但它还将向调用堆栈添加一个额外的调用。因此，我的堆栈错误跟踪如下所示：

...
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/blocking_connection.py", line 1336, in _flush_output
    self._connection._flush_output(lambda: self.is_closed, *waiters)
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/blocking_connection.py", line 522, in _flush_output
    raise self._closed_result.value.error
pika.exceptions.StreamLostError: Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/dist-packages/msworker/queue.py", line 81, in listen
    self.channel.basic_ack(delivery_tag=method_frame.delivery_tag)
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/blocking_connection.py", line 2113, in basic_ack
    self._flush_output()
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/blocking_connection.py", line 1336, in _flush_output
    self._connection._flush_output(lambda: self.is_closed, *waiters)
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/blocking_connection.py", line 522, in _flush_output
    raise self._closed_result.value.error
pika.exceptions.StreamLostError: Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/dist-packages/msworker/queue.py", line 81, in listen
    self.channel.basic_ack(delivery_tag=method_frame.delivery_tag)
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/blocking_connection.py", line 2113, in basic_ack
    self._flush_output()
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/blocking_connection.py", line 1336, in _flush_output
    self._connection._flush_output(lambda: self.is_closed, *waiters)
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/blocking_connection.py", line 522, in _flush_output
    raise self._closed_result.value.error
pika.exceptions.StreamLostError: Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/utils/io_services_utils.py", line 1097, in _on_socket_writable
    self._produce()
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/utils/io_services_utils.py", line 820, in _produce
    self._tx_buffers[0])
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/utils/io_services_utils.py", line 79, in retry_sigint_wrap
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.6/dist-packages/pika/adapters/utils/io_services_utils.py", line 861, in _sigint_safe_send
    return sock.send(data)
ConnectionResetError: [Errno 104] Connection reset by peer

在调用堆栈中没有旧调用的情况下，如何从头开始重新运行listen函数

更新

为避免此问题，可以操作嵌套函数并重新运行它，但不能重新运行它本身：

def listen(self, pipeline):
    try:
        self._listen(self, pipeline)
    except (pika.exceptions.StreamLostError,
            pika.exceptions.ConnectionClosed,
            pika.exceptions.ChannelClosed,
            ConnectionResetError) as e:
        logging.warning(f'Connection dropped for queue {self.queue_name}. Exception: {e}. Reconnecting...')
        self._reconnect()
        self._listen(self, pipeline)


def _listen(self, pipeline):
    for message in self.channel.consume(self.queue_name):
            pipeline.process(message)

但是，有没有办法用干净的调用堆栈重新运行递归函数

Answer 1

既然可以使用简单迭代，为什么还要使用递归呢

def listen(self, pipeline):
    while True:
        try:
            for message in self.channel.consume(self.queue_name):
                pipeline.process(message)
                self.channel.basic_ack(delivery_tag=method_frame.delivery_tag)
            return

        except (pika.exceptions.StreamLostError,
                pika.exceptions.ConnectionClosed,
                pika.exceptions.ChannelClosed,
                ConnectionResetError) as e:
            logging.warning(f'Connection dropped for queue {self.queue_name}. Exception: {e}. Reconnecting...')
        self._reconnect()

But still, is there a way to rerun the recursive function with a clean call stack?

实际上，您当前拥有的是一个“干净的调用堆栈”-它是real调用堆栈，每个调用有一个不同的帧（递归或非递归）。有些语言确实“优化”了尾部递归调用（通过挤压/重用帧），Python的设计者选择不让调试更容易