使用Tornado将二进制文件流式传输到Google Storage

Question

我正在尝试通过我的服务器将二进制文件从客户端请求流式传输到Google云端存储.

我正在使用Tornado框架将请求中的数据流式传输到服务器,并使用Google Cloud Storage API将文件流式传输到Google upload_from_file方法.

我是Tornado的新手,我正在使用@stream_request_body装饰器,所以我可以从块中获取请求中的数据并将每个块上传到Google.

我试图打开一个类似文件的对象,我可以将每个块写入,同时将"文件"本身上传到Google.

问题是我在开始编写块之前无法将"文件"上传到Google.

任何援助将不胜感激.

Answer 1

使用 Google 的 HTTP 库执行此操作很棘手，因为它们是为同步使用而设计的。您需要将实际上传放在另一个线程上以避免阻塞 IOLoop。您可以使用os.pipe来在 Tornado 线程和上传线程之间进行通信（将管道的写入端包装在 PipeIOStream 中，将读取端包装在 中os.fdopen）。这是未经测试的解决方案草图：

def prepare(self):
    r, w = os.pipe()
    self.write_pipe = tornado.iostream.PipeIOStream(w)
    # Create our "file-like object" for upload_from_file
    self.read_pipe = os.fdopen(r)
    # Create an event for the upload thread to communicate back
    # to tornado when it's done, and save a reference to our IOLoop.
    self.upload_done = tornado.locks.Event()
    self.io_loop = tornado.ioloop.IOLoop.current()
    # Consider using a tornado.locks.Semaphore to limit the number of
    # threads you can create.
    self.thread = threading.Thread(target=self.upload_file)
    self.thread.start()

def upload_file(self):
    google_client.upload_from_file(self.read_pipe)
    # tell the IOLoop thread we're finished
    self.io_loop.add_callback(self.upload_done.set)

async def data_received(self, chunk):
    await self.write_pipe.write(chunk)

async def put(self):  # or post()
    self.write_pipe.close()
    await self.upload_done.wait()
    self.thread.join()
    self.render("upload_done.html")

或者，您可以避免使用 google 的同步库，并使用底层 HTTP API 和 AsyncHTTPClient 来完成所有操作。以这种方式排序身份验证很棘手，但可以避免线程不匹配。这将涉及使用 body_ Producer ，如本要点所示