Tornado 非同步讀寫文件的方法？

大家好，

我現在正在使用Tornado開發一個文件上傳應用。當通過Tornado主線程非同步獲取到用戶上傳的文件數據後，我需要把數據寫入到文件系統中。

1. 如果為每個文件的寫操作單獨開一個線程/進程，由於可能同時寫很多文件，大量的線程/進程對伺服器會是很大的開銷，而且Tornado也不建議亂開線程/進程；

2. 如果把所有待寫入的數據放到隊列中，然後用一個線程周期性地從隊列中獲取數據並寫入到文件系統中，則會涉及到對隊列的加鎖問題。而鎖的引入勢必又會拖慢Tornado的速度（因為Tornado必須獲取鎖後才能把數據寫到隊列中）。

3. 一個更理想的做法是採用類似Tornado讀寫socket的epoll模型，這樣既可以不開線程/進程，又不會阻塞，但普通文件的讀寫又不支持epoll（即不能設置為non-block）。

所以想請問大家，像這種非同步讀寫文件的問題一般是怎麼解決的呢？謝謝！

---

這個系統用tornado搭建本身就選擇錯誤。原因如下：

1. 對於讀取上傳文件，tornado的官網上有說過，對於文件上傳，tornado是直接全部讀到內存中，再進行處理，也就是說，文件越大，文件越多，你的資源佔用就越多。官網推薦用 nginx 的upload module 來處理

2. 寫文件這個操作，如果你用的是機械磁碟，本身就是同步的。因為只有一個磁頭，每次的定址都是同步的。所以，tornado的寫文件操作無法非同步。

如果一定用tornado，需要這麼做：

1. 整個系統用nginx+tornado [+另外的server]

2. 對於文件上傳，採用nginx upload module，nginx負責文件的上傳，tornado只負責進行文件的cp和mv

3. 如果你的文件上傳後有後續複雜的io操作。比方說：讀取文件裡面的內容，找出之前的某個相關文件，並且合併，再重新寫入。那麼需要一個另外的server，專門負責這個。

4. 對於第3點，如果你採用非機械磁碟，應該就沒有必要了。

---

2. 如果把所有待寫入的數據放到隊列中，然後用一個線程周期性地從隊列中獲取數據並寫入到文件系統中，則會涉及到對隊列的加鎖問題。而鎖的引入勢必又會拖慢Tornado的速度（因為Tornado必須獲取鎖後才能把數據寫到隊列中）。

這個不會慢的，鎖這種原語在 ns 級別，寫文件在 ms+ 級別。

可以考慮現成的隊列系統，SDK 一般都包含 Python。主流的隊列 Worker 也不會採用周期讀取的策略，而是消息通知方式。

---

如果僅僅是上傳文件，相關的邏輯很少，使用nginx+module處理就很好

如果上傳文件還有較多的邏輯處理，nodejs的非阻塞io也非常好用。

如果一定要用torado，那麼加鎖的開銷其實不大，每秒進行lock十萬次應該是沒問題的

---

Tornado 真是io傷不起啊, 一旦阻塞整個服務都block..

所以耗時的任務採用非同步httpclient交給後端server處理..或採用專業的MQ解決你說的鎖問題

---

嘎嘎，原來貌似只要給類加上stream_request_body裝飾器，data_received就是非同步的了，只管在data_received裡面進行寫文件就好了

==================old================

實在沒辦法，nginx的upload_module在高版本編不過去，網上又找不到可用的py庫，我就這樣了

?⊙?⊙?

至少是沒堵著。。。

from tornado.web import RequestHandler
from tornado import gen
from tornado.web import stream_request_body
import time
import random

@stream_request_body
class A(RequestHandler):
def initialize(self):
self.fp = open("xxxx.tmp","wb")
self.size = 0
self.id = random.randint(0,100)

async def post(self):
self.set_header("Content-Type", "application/json;charset=utf-8")
self.finish("{"msg":"ok"}")
await gen.sleep(0)
self.fp.close()

async def data_received(self, chunk):
self.size += len(chunk)
print(time.time(), self.id, self.size)
self.fp.write(chunk)
await gen.sleep(0)


---

tornado-celery

好像挖到一個老坑

---

2，用現成的 MQ 如 Rabbit 之類。

另外，如果你的 I/O 速度成瓶頸，前端響應客戶端請求速度再快也沒有意義。

---

推薦閱讀：