回答此问题可获得 20 贡献值,回答如果被采纳可获得 50 分。
<p>在使用<code>asyncio</code>时,从信号处理程序回调引发自定义异常时,我在捕获该异常时遇到问题</p>
<p>如果我从下面的<code>do_io()</code>中抛出<code>ShutdownApp</code>,我就能够在<code>run_app()</code>中正确地捕捉到它。然而,当从<code>handle_sig()</code>引发异常时,我似乎无法捕获它</p>
<p><a href="https://stackoverflow.com/help/minimal-reproducible-example">Minimal, Reproducible Example</a>使用Python 3.8.5进行测试:</p>
<pre class="lang-py prettyprint-override"><code>import asyncio
from functools import partial
import os
import signal
from signal import Signals
class ShutdownApp(BaseException):
pass
os.environ["PYTHONASYNCIODEBUG"] = "1"
class App:
def __init__(self):
self.loop = asyncio.get_event_loop()
def _add_signal_handler(self, signal, handler):
self.loop.add_signal_handler(signal, handler, signal)
def setup_signals(self) -> None:
self._add_signal_handler(signal.SIGINT, self.handle_sig)
def handle_sig(self, signum):
print(f"\npid: {os.getpid()}, Received signal: {Signals(signum).name}, raising error for exit")
raise ShutdownApp("Exiting")
async def do_io(self):
print("io start. Press Ctrl+C now.")
await asyncio.sleep(5)
print("io end")
def run_app(self):
print("Starting Program")
try:
self.loop.run_until_complete(self.do_io())
except ShutdownApp as e:
print("ShutdownApp caught:", e)
# TODO: do other shutdown related items
except:
print("Other error")
finally:
self.loop.close()
if __name__ == "__main__":
my_app = App()
my_app.setup_signals()
my_app.run_app()
print("Finished")
</code></pre>
<p>在异步调试模式下按<code>CTRL+C</code>(对于<code>SIGINT</code>)后的输出:</p>
<pre><code>(env_aiohttp) anav@anav-pc:~/Downloads/test$ python test_asyncio_signal.py
Starting Program
io start. Press Ctrl+C now.
^C
pid: 20359, Received signal: SIGINT, raising error for exit
Exception in callback App.handle_sig(<Signals.SIGINT: 2>)
handle: <Handle App.handle_sig(<Signals.SIGINT: 2>) created at /home/anav/miniconda3/envs/env_aiohttp/lib/python3.8/asyncio/unix_events.py:99>
source_traceback: Object created at (most recent call last):
File "test_asyncio_signal.py", line 50, in <module>
my_app.setup_signals()
File "test_asyncio_signal.py", line 25, in setup_signals
self._add_signal_handler(signal.SIGINT, self.handle_sig)
File "test_asyncio_signal.py", line 22, in _add_signal_handler
self.loop.add_signal_handler(signal, handler, signal)
File "/home/anav/miniconda3/envs/env_aiohttp/lib/python3.8/asyncio/unix_events.py", line 99, in add_signal_handler
handle = events.Handle(callback, args, self, None)
Traceback (most recent call last):
File "/home/anav/miniconda3/envs/env_aiohttp/lib/python3.8/asyncio/events.py", line 81, in _run
self._context.run(self._callback, *self._args)
File "test_asyncio_signal.py", line 31, in handle_sig
raise ShutdownApp("Exiting")
ShutdownApp: Exiting
io end
Finished
</code></pre>
<p>预期产出:</p>
<pre><code>Starting Program
io start. Press Ctrl+C now.
^C
pid: 20359, Received signal: SIGINT, raising error for exit
ShutdownApp caught: Exiting
io end
Finished
</code></pre>
<p>是否可以从<code>asyncio</code>中的信号处理程序引发自定义异常?如果是这样,我如何正确地捕捉/排除它</p>