我有一个使用asyncio
的协同线方法的事件循环。
我热衷于使用uvloop来寻找以下类似的示例。
下面是一个简单的asyncio
事件循环示例:
import asyncio
async def read(**kwargs):
oid = kwargs.get('oid', '0.0.0.0.0.0')
time = kwargs.get('time', 1)
try:
print('start: ' + oid)
except Exception as exc:
print(exc)
finally:
await asyncio.sleep(time)
print('terminate: ' + oid)
def event_loop(configs):
loop = asyncio.get_event_loop()
for conf in configs:
asyncio.ensure_future(read(oid=conf['oid'], time=conf['time']))
return loop
if __name__ == '__main__':
snmp_configurations = [
{'time': 5, 'oid': '1.3.6.3.2.4'},
{'time': 6, 'oid': '1.3.6.3.5.8'},
] # TODO :: DUMMY
loop = event_loop(snmp_configurations)
try:
loop.run_forever()
except KeyboardInterrupt:
pass
finally:
print("Closing Loop")
loop.close()
问题:
注意事项
发布于 2018-11-03 10:02:28
只需在调用asyncio.get_event_loop()
之前设置事件循环策略即可。
import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
async def read(**kwargs):
oid = kwargs.get('oid', '0.0.0.0.0.0')
time = kwargs.get('time', 1)
try:
print('start: ' + oid)
except Exception as exc:
print(exc)
finally:
await asyncio.sleep(time)
print('terminate: ' + oid)
def event_loop(configs):
loop = asyncio.get_event_loop()
for conf in configs:
asyncio.ensure_future(read(oid=conf['oid'], time=conf['time']))
return loop
if __name__ == '__main__':
snmp_configurations = [
{'time': 5, 'oid': '1.3.6.3.2.4'},
{'time': 6, 'oid': '1.3.6.3.5.8'},
] # TODO :: DUMMY
loop = event_loop(snmp_configurations)
try:
loop.run_forever()
except KeyboardInterrupt:
pass
finally:
print("Closing Loop")
loop.close()
是的,这个代码是正确的。可以在导入后设置事件循环策略。
import uvloop
import asyncio
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) # TODO :: uvloop.
def event_loop(configs):
loop = asyncio.get_event_loop()
for conf in configs:
asyncio.ensure_future(read(oid=conf['oid'], time=conf['time']))
return loop
https://stackoverflow.com/questions/53130168
复制相似问题