danmuji = bilibiliClient(url, self.lock, self.commentq, self.numq)
task1 = asyncio.ensure_future(danmuji.connectServer())
task2 = asyncio.ensure_future(danmuji.HeartbeatLoop())

其實若將心跳任務 HeartbeatLoop 放入 connectorServer 中去啟動，代碼看起來更優(yōu)雅一些。但這么做是因為我需要維護一個任務列表，后面會有描述。

在彈幕存儲上我花了些時間選擇。

數(shù)據(jù)庫存儲是一個同步 IO 的過程，Insert 的時候會阻塞彈幕收集的任務。雖然有 aiomysql 這種異步接口，但配置數(shù)據(jù)庫太麻煩，我的設想是這個小系統(tǒng)能夠方便地部署。

最終我選擇使用自帶的 sqlite3。但 sqlite3 無法做并行操作，故開了一個線程單獨進行數(shù)據(jù)庫存儲。在另一個線程中，100 * 2 個任務搜集所有的彈幕、人數(shù)信息，并塞進隊列 commentq, numq 中。存儲線程每隔 10s 喚醒一次，將隊列中的數(shù)據(jù)寫進 sqlite3 中，并清空隊列。

在多線程和異步的配合下，網(wǎng)絡流量沒有被阻塞。

可能的連接失敗場景處理

彈幕協(xié)議是直接基于 TCP，位與位直接關聯(lián)性較強，一旦解析錯誤，很容易就拋 Exception（個人感覺，雖然 TCP 是可靠傳輸，但B站服務器自身發(fā)生錯誤也是有可能的）。所以有必要設計一個自動重連機制。

在 asyncio 文檔中提到，

Done means either that a result / exception are available, or that the future was cancelled.

函數(shù)正常返回、拋出異?；蛘呤潜?cancel，都會退出當前任務。可以使用 done() 來判斷。

每一個直播間對應兩個任務，解析任務是最容易掛的，但并不會影響心跳任務，所以必須找出并將對應心跳任務結束。
在創(chuàng)建任務的時候使用字典記錄每個房間的兩個任務，

self.tasks[url] = [task1, task2]

在運行過程中，每隔 10s 做一次檢查，

for url in self.tasks:
  item = self.tasks[url]
  task1 = item[0]
  task2 = item[1]
  if task1.done() == True or task2.done() == True:
    if task1.done() == False:
      task1.cancel()
    if task2.done() == False:
      task2.cancel()
    danmuji = bilibiliClient(url, self.lock, self.commentq, self.numq)
    task11 = asyncio.ensure_future(danmuji.connectServer())
    task22 = asyncio.ensure_future(danmuji.HeartbeatLoop())
    self.tasks[url] = [task11, task22]

實際我只見過一次任務失敗的場景，是因為主播房間被封了，導致無法進入直播間。

結論