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

Question

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

我正在使用 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，如 this gist