如何仅用asyncio模块在异步函数中读取JSON文件？

核心结论 #

不是必须使用aiofiles，仅用asyncio完全可以完成异步环境下的JSON文件读写操作，具体实现分两种场景选择：

场景1：简单脚本（无其他并发异步任务） #

如果你的程序仅处理这一个文件操作，没有其他需要并发执行的异步任务，直接使用同步文件操作最简洁，无需额外依赖：

import asyncio
import json

async def main():
    # 同步读取JSON文件并直接解析为字典
    with open('rhydon.json', mode='r') as f:
        pokemon = json.load(f)
    
    name = pokemon['name']
    moves = [move['move']['name'] for move in pokemon['moves']]
    
    # 同步写入结果文件
    with open(f'{name}_moves.txt', mode='w') as f:
        f.write('\n'.join(moves))

asyncio.run(main())

这种写法省去了aiofiles的引入和异步读取的繁琐步骤，但要注意：同步IO会阻塞整个async事件循环——如果程序只有这一个任务，完全不影响；但如果有其他异步任务（比如同时处理多个文件、网络请求），阻塞会导致其他任务延迟。

场景2：大型异步程序（需避免事件循环阻塞） #

如果你的程序需要同时处理多个异步任务，为了不阻塞事件循环，可以用asyncio.run_in_executor将同步文件操作丢到线程池执行，原理和aiofiles一致，但无需额外安装依赖：

import asyncio
import json
from concurrent.futures import ThreadPoolExecutor

async def main():
    with ThreadPoolExecutor() as executor:
        # 异步读取JSON文件（线程池执行同步操作）
        pokemon = await asyncio.get_event_loop().run_in_executor(
            executor,
            lambda: json.load(open('rhydon.json', 'r'))
        )
        
        name = pokemon['name']
        moves_str = '\n'.join([move['move']['name'] for move in pokemon['moves']])
        
        # 异步写入文件（线程池执行同步操作）
        await asyncio.get_event_loop().run_in_executor(
            executor,
            lambda: open(f'{name}_moves.txt', 'w').write(moves_str)
        )

asyncio.run(main())

总结 #

• 仅用asyncio实现需求完全可行，不需要aiofiles
• 简单场景下直接用同步IO最简洁
• 并发场景下用run_in_executor是更优的非阻塞方案

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