将aiohttp脚本转为asyncio+requests（Ubuntu下aiohttp无法运行）

问题背景 #

原aiohttp脚本在macOS上可正常运行，但Ubuntu环境下aiohttp无法工作。经测试，单独使用requests或简单的asyncio+requests组合能正常访问目标网站，于是将原脚本改为requests版本，却出现TypeError循环报错。

原aiohttp代码 #

import aiohttp
import asyncio

async def get_data(session, x):
    while True:
        try:
            async with session.get(url=f'https://api.abc.com/{x}') as response:
                if response.status == 200:
                    data = await response.json()
                    try:
                        data = float(data)
                        return data
                    except ValueError:
                        print("Data is not a valid float. Retrying...")
                else:
                    print("Received non-200 status code. Retrying...")
            await asyncio.sleep(1)  # Wait for 1 second before retrying
        except Exception as e:
            print("Unable to get url {} due to {}. Retrying...".format(x, e.__class__))
            await asyncio.sleep(1)  # Wait for 1 second before retrying

async def main(datas):
    async with aiohttp.ClientSession() as session:
        ret = await asyncio.gather(*[get_data(session, data) for data in datas])
        return {datas[i]: ret[i] for i in range(len(datas))}  # Return the results as a dictionary

datas = ['x1', 'x2', 'x3', 'x4']
results = asyncio.run(main(datas))

修改后的requests代码 #

import asyncio
import requests

async def get_data(x):
    while True:
        try:
            response = requests.get(url=f'https://api.abc.com/{x}')
            if response.status_code == 200:
                try:
                    data = float(response.json())
                    return data
                except ValueError:
                    print("Data is not a valid float. Retrying...")
            else:
                print("Received non-200 status code. Retrying...")
            await asyncio.sleep(1)  # Wait for 1 second before retrying
        except Exception as e:
            print("Unable to get url {} due to {}. Retrying...".format(x, e.__class__))
            await asyncio.sleep(1)  # Wait for 1 second before retrying

async def main(datas):
    tasks = [get_data(data) for data in datas]
    ret = await asyncio.gather(*tasks)
    return {datas[i]: ret[i] for i in range(len(datas))}  # Return the results as a dictionary

datas = ['x1', 'x2', 'x3', 'x4']
results = asyncio.run(main(datas))
print(results)

报错信息 #

Unable to get data for x1 due to <class 'TypeError'>. Retrying...
Unable to get data for x2 due to <class 'TypeError'>. Retrying...
Unable to get data for x3 due to <class 'TypeError'>. Retrying...
Unable to get data for x4 due to <class 'TypeError'>. Retrying...
Unable to get data for x1 due to <class 'TypeError'>. Retrying...
...

解决方案 #

问题根源是requests.get是同步阻塞函数，在asyncio协程中直接调用会堵塞事件循环，导致异常捕获逻辑紊乱。正确做法是用asyncio.to_thread将同步的requests调用放到线程池执行，避免堵塞事件循环。

修改后的正确代码：

import asyncio
import requests

async def get_data(x):
    while True:
        try:
            # 用to_thread将同步请求包装为异步操作
            response = await asyncio.to_thread(requests.get, url=f'https://api.abc.com/{x}')
            if response.status_code == 200:
                try:
                    data = float(response.json())
                    return data
                except ValueError:
                    print("数据不是有效的浮点数，重试中...")
            else:
                print("收到非200状态码，重试中...")
            await asyncio.sleep(1)
        except Exception as e:
            print(f"无法获取{x}的数据，错误类型：{type(e)}，重试中...")
            await asyncio.sleep(1)

async def main(datas):
    tasks = [get_data(data) for data in datas]
    ret = await asyncio.gather(*tasks)
    return {datas[i]: ret[i] for i in range(len(datas))}

datas = ['x1', 'x2', 'x3', 'x4']
results = asyncio.run(main(datas))
print(results)

版本兼容说明 #

如果你的Python版本低于3.9（asyncio.to_thread是3.9新增特性），可以用loop.run_in_executor替代：

loop = asyncio.get_running_loop()
response = await loop.run_in_executor(None, requests.get, f'https://api.abc.com/{x}')

内容的提问来源于stack exchange，提问作者anarchy