为何调用asyncio.get_event_loop时出现‘DeprecationWarning: 无当前事件循环’警告？

运行以下Python代码时，第35行的loop = asyncio.get_event_loop()语句触发DeprecationWarning，警告内容为：

LXPReadTester.py:35: DeprecationWarning: There is no current event loop
  loop = asyncio.get_event_loop()

代码示例 #

import sys

from LXPPacket import LXPPacket
import asyncio

SERVER_URL = "172.16.255.120"
# SERVER_URL = "127.0.0.1"
SERVER_PORT = 8000


async def tcp_echo_client(loop):
    while True:
        try:
            address = int(input("Register Address: "))
            reader, writer = await asyncio.open_connection(SERVER_URL, SERVER_PORT,
                                                           loop=loop)
            print("Connected to server")
            lxpPacket = LXPPacket()
            packet = lxpPacket.prepare_packet_for_read(address, 1)
            writer.write(packet)
            await writer.drain()
            # await asyncio.sleep(1)
            packet = await reader.read(1000)
            print('Received: ', packet)
            result = lxpPacket.parse_packet(packet)
            if not lxpPacket.packet_error:
                print(result)
            print('Close the socket')
            writer.close()
        except Exception as e:
            print("Exception ", e)
            print(sys.gettrace())

if __name__ =="__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(tcp_echo_client(loop))
    loop.close()

警告产生原因 #

在Python 3.10及以上版本中，asyncio.get_event_loop()的行为发生了变更：

• 旧版本中，如果当前线程没有已存在的事件循环，该方法会自动创建一个新的事件循环并设置为当前循环。
• 新版本中，这种自动创建的行为被标记为废弃，当没有当前事件循环时调用get_event_loop()就会触发该警告，后续Python版本会直接抛出RuntimeError而不是自动创建循环。

解决方法 #

方法1：手动创建并设置事件循环 #

如果需要保留手动管理循环的方式，可以改用asyncio.new_event_loop()创建新循环，再通过asyncio.set_event_loop()将其设为当前循环：

if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(tcp_echo_client(loop))
    loop.close()

方法2：使用官方推荐的asyncio.run()（更简洁） #

Python 3.7+提供了asyncio.run()方法，它会自动完成事件循环的创建、运行和关闭，是官方推荐的启动asyncio程序的方式。需要对代码做两处调整：

1. 修改tcp_echo_client函数，移除不需要的loop参数，同时去掉asyncio.open_connection中的loop参数（新版本会自动使用当前循环）：

async def tcp_echo_client():
    while True:
        try:
            address = int(input("Register Address: "))
            reader, writer = await asyncio.open_connection(SERVER_URL, SERVER_PORT)
            # 其余代码保持不变...

2. 修改main函数，用asyncio.run()启动任务：

if __name__ == "__main__":
    asyncio.run(tcp_echo_client())

这种方式无需手动管理事件循环，代码更简洁，也能避免版本兼容问题。

内容的提问来源于stack exchange，提问作者Dion Van den Berg