对一个爬虫的异步函数，在其功能函数上添加装饰器，用于打印错误日志
import aiohttp
import asyncio
import run_change # 这是一个防止 asyncio.run()有时会报错 RuntimeError:Event loop is closed 的模块，需要手动编写，下面一个代码块有提供
def wrapper1(func):
async def wrapper(*args, err, **kwargs):
result = await asyncio.create_task(func(*args, **kwargs)) # 执行传入的函数
# print(result)
if not result:
print(err)
return result
return wrapper
@wrapper1
async def get_message(session):
# print('开始访问')
async with session.get('https://www.baidu.com') as response:
# print('访问结束')
return await response.text()
async def run():
async with aiohttp.ClientSession() as session:
return await asyncio.create_task(get_message(session, err='not')) # 将功能函数加入任务列表
if __name__ == '__main__':
s = asyncio.run(run())
# run_change.py
"""
【解决 asyncio.run()有时会报错 RuntimeError:Event loop is closed 情况】
当asyncio.run()被修改为:
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
确实不会报错
大佬总结, 总而言之是asyncio.run()会自动关闭循环,并且调用_ProactorbasePipeTransport.__del__报错, 而asyncio.run_until_complete()不会.
大佬提供了解决方案, 就是重写方法以保证run()的运行.
【导入此包即可解决问题，如若还不能请使用上面的方式】
"""
from functools import wraps
from asyncio.proactor_events import _ProactorbasePipeTransport
def silence_event_loop_closed(func):
@wraps(func)
def wrapper(self, *args, **kwargs):
try:
return func(self, *args, **kwargs)
except RuntimeError as e:
if str(e) != 'Event loop is closed':
raise
return wrapper
_ProactorbasePipeTransport.__del__ = silence_event_loop_closed(_ProactorbasePipeTransport.__del__)
