复用aiobotocore客户端异步上传S3时遇‘cannot reuse already awaited queue’错误

错误原因 #

你遇到的cannot reuse already awaited queue错误，本质是因为aiobotocore的客户端实例通过async with上下文管理器使用时，退出上下文会自动关闭内部的异步任务队列、连接池等资源。这些资源一旦被关闭就无法再次初始化，所以复用同一个客户端实例再次进入async with上下文就会触发该错误。

解决方案 #

方案1：移除async with直接复用客户端 #

aiobotocore的客户端实例本身支持安全复用，无需每次通过上下文管理器包裹调用。直接调用客户端的方法即可，仅在不再需要客户端时手动关闭资源：

修改调用代码：

await self.client.put_object(
    Bucket=self.bucket_name,
    Key=self.key_name,
    Body=BytesIO(self.compressed_image)
)

为了在类实例销毁时正确释放资源，可以给S3Bucket类添加异步关闭方法：

class S3Bucket:
    # __init__方法保持原有配置不变
    
    async def close(self):
        """手动关闭客户端释放资源"""
        await self.client.close()

在程序结束或不再需要该实例时，调用await s3_bucket_instance.close()即可完成资源清理。

方案2：复用会话，按需创建客户端（备选） #

如果必须使用上下文管理器模式，可以复用aiobotocore.Session实例，每次从会话创建新客户端（会话复用的开销远低于重新创建会话）：

修改类的初始化代码：

class S3Bucket:
    def __init__(self) -> None:
        # 其他配置项保持不变
        self.session = get_session()  # 复用会话实例
        self.client = None
        
    async def get_client(self):
        """按需创建/复用客户端"""
        if not self.client or self.client._closed:
            self.client = self.session.create_client(
                's3',
                aws_access_key_id=self.access_key,
                aws_secret_access_key=self.secret_key,
                region_name=self.bucket_region
            )
        return self.client

调用时：

async with await self.get_client() as s3_client:
    await s3_client.put_object(
        Bucket=self.bucket_name,
        Key=self.key_name,
        Body=BytesIO(self.compressed_image)
    )

但注意，这种方式每次创建客户端仍会有一定开销，方案1是更优选择，既满足复用需求又避免重复创建客户端的耗时。

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