如何同时使用io模块与tornado.write？Tornado代理谷歌云盘大文件问题

嘿，我之前正好解决过类似的场景——用Tornado做HTTP代理，分块传输谷歌云盘的大文件，而MediaIoBaseDownload只支持IO类对象的限制确实有点棘手，不过我们可以通过自定义IO类来绕开这个问题，直接把下载的分块数据推送给客户端，不用先缓存到本地或内存里。

核心思路 #

MediaIoBaseDownload需要一个实现了write方法的IO对象来接收分块数据，那我们就自定义一个IO类，让它的write方法直接调用Tornado的self.write()和self.flush()，把下载到的每一块数据实时传给客户端。同时，为了不阻塞Tornado的异步IO循环，要把同步的下载逻辑放到线程池里执行。

具体实现代码 #

1. 自定义Tornado分块写入IO类

这个类继承自io.IOBase，负责把谷歌云盘下载的分块直接转发给客户端：

import io
from tornado.web import RequestHandler

class TornadoChunkWriter(io.IOBase):
    def __init__(self, handler: RequestHandler):
        self.handler = handler
        # 分块大小可根据需求调整，建议1MB~10MB
        self.chunk_size = 1024 * 1024  

    def write(self, b):
        # 将下载到的二进制分块直接写入响应
        self.handler.write(b)
        # 立即flush确保数据发送到客户端，避免积压
        self.handler.flush()
        # 返回写入的字节数，符合IO类的要求
        return len(b)

    def flush(self):
        # 重写flush方法，同步Tornado的响应刷新
        self.handler.flush()

2. Tornado请求处理器实现

在请求处理器中，我们初始化MediaIoBaseDownload，用自定义的IO类接收数据，并通过线程池执行同步下载逻辑：

from pydrive2.drive import GoogleDrive
from googleapiclient.http import MediaIoBaseDownload
import tornado.ioloop
import tornado.web

class GDriveProxyHandler(tornado.web.RequestHandler):
    async def get(self):
        # 初始化PyDrive的GoogleDrive实例（确保已完成认证）
        drive = GoogleDrive()
        # 从请求参数获取云盘文件ID，你也可以根据需求调整获取方式
        file_id = self.get_query_argument("file_id")
        gdrive_file = drive.CreateFile({'id': file_id})

        # 设置响应头，让客户端正确识别文件
        self.set_header("Content-Type", gdrive_file['mimeType'])
        self.set_header("Content-Length", gdrive_file['fileSize'])
        self.set_header(
            "Content-Disposition",
            f'attachment; filename="{gdrive_file["title"]}"'
        )

        # 创建自定义的分块写入器
        chunk_writer = TornadoChunkWriter(self)

        # 初始化MediaIoBaseDownload
        download_request = drive.auth.service.files().get_media(fileId=file_id)
        downloader = MediaIoBaseDownload(
            chunk_writer,
            download_request,
            chunksize=chunk_writer.chunk_size
        )

        try:
            # 将同步的下载循环放到线程池执行，避免阻塞Tornado主IO循环
            await tornado.ioloop.IOLoop.current().run_in_executor(
                None,
                self._execute_download,
                downloader
            )
        except Exception as e:
            # 捕获异常，返回500错误
            self.send_error(500, reason=str(e))

    def _execute_download(self, downloader):
        done = False
        while not done:
            status, done = downloader.next_chunk()
            # 可选：打印下载进度日志
            # print(f"传输进度: {status.progress() * 100:.2f}%")

# 初始化Tornado应用
def make_app():
    return tornado.web.Application([
        (r"/gdrive-proxy", GDriveProxyHandler),
    ])

if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    tornado.ioloop.IOLoop.current().start()

关键注意事项 #

• 异步线程池的使用：MediaIoBaseDownload的next_chunk()是同步阻塞方法，必须放到Tornado的线程池里执行，否则会卡住整个服务器的IO循环，影响其他请求。
• 响应头设置：一定要正确设置Content-Length和Content-Type，这样客户端才能显示下载进度并正确处理文件；Content-Disposition可以控制文件是直接下载还是在线预览。
• 错误处理：添加异常捕获逻辑，避免下载过程中出现错误导致服务器崩溃，同时给客户端返回明确的错误信息。
• 认证问题：确保你的PyDrive实例已经完成了谷歌云盘的认证（比如通过服务账号OAuth2），否则无法访问目标文件。

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