把酒言歡話聊天，基於Vue3.0+Tornado6.1+Redis發佈訂閲(pubsub)模式非阻塞實時通信聊天系統詳情 - tornado,redis,websocket,通信協議,實時通信劉悦的技術博客動態日志

原文轉載自「劉悦的技術博客」https://v3u.cn/a_id_202

“表達欲”是人類成長史上的強大“源動力”，恩格斯早就直截了當地指出，處在矇昧時代即低級階段的人類，“以果實、堅果、根作為食物；音節清晰的語言的產生是這一時期的主要成就”。而在網絡時代人們的表達欲往往更容易被滿足，因為有聊天軟件的存在。通常意義上，聊天大抵都基於兩種形式：羣聊和單聊。羣聊或者羣組聊天我們可以理解為聊天室，可以有人數上限，而單聊則可以認為是上限為2個人的特殊聊天室。

為了開發高質量的聊天系統，開發者應該具備客户機和服務器如何通信的基本知識。在聊天系統中，客户端可以是移動應用程序(C端)或web應用程序(B端)。客户端之間不直接通信。相反，每個客户端都連接到一個聊天服務，該服務支撐雙方通信的功能。所以該服務在業務上必須支持的最基本功能：

1.能夠實時接收來自其他客户端的信息。

2.能夠將每條信息實時推送給收件人。

當客户端打算啓動聊天時，它會使用一個或多個網絡協議連接聊天服務。對於聊天服務，網絡協議的選擇至關重要，這裏，我們選擇Tornado框架內置Websocket協議的接口，簡單而又方便，安裝tornado6.1

pip3 install tornado==6.1

隨後編寫程序啓動文件main.py:

import tornado.httpserver  
import tornado.websocket  
  
import tornado.ioloop  
  
import tornado.web  
  
import redis  
  
import threading  
  
import asyncio  
  
# 用户列表  
users = []  
  
# websocket協議  
class WB(tornado.websocket.WebSocketHandler):  
  
  
    # 跨域支持  
    def check_origin(self,origin):  
  
        return True  
  
    # 開啓鏈接  
    def open(self):  
  
                users.append(self)  
  
  
    # 接收消息  
    def on_message(self,message):  
  
        self.write_message(message['data'])  
  
    # 斷開  
    def on_close(self):  
  
        users.remove(self)

# 建立torando實例  
  
app = tornado.web.Application(  
  
    [  
  
    (r'/wb/',WB)  
  
    ],debug=True  
  
)  
  
if __name__ == '__main__':  
  
  
    # 聲明服務器  
    http_server_1 = tornado.httpserver.HTTPServer(app)  
  
    # 監聽端口  
    http_server_1.listen(8000)  
  
    # 開啓事件循環  
    tornado.ioloop.IOLoop.instance().start()

如此，就在短時間搭建起了一套websocket協議服務，每一次有客户端發起websocket連接請求，我們都會將它添加到用户列表中，等待用户的推送或者接收信息的動作。

下面我們需要通過某種形式將消息的發送方和接收方聯繫起來，以達到“聊天”的目的，這裏選擇Redis的發佈訂閲模式(pubsub)，以一個demo來實例説明，server.py

import redis  
  
r = redis.Redis()  
r.publish("test",'hello')

隨後編寫 client.py:

import redis  
r = redis.Redis()  
ps = r.pubsub()  
ps.subscribe('test')    
for item in ps.listen():   
    if item['type'] == 'message':  
        print(item['data'])

可以這麼理解：訂閲者(listener)負責訂閲頻道(channel)；發送者(publisher)負責向頻道(channel)發送二進制的字符串消息，然後頻道收到消息時，推送給訂閲者。

頻道不僅可以聯繫發佈者和訂閲者，同時，也可以利用頻道進行“消息隔離”，即不同頻道的消息只會給訂閲該頻道的用户進行推送：

根據發佈者訂閲者邏輯，改寫main.py:

import tornado.httpserver  
import tornado.websocket  
  
import tornado.ioloop  
  
import tornado.web  
  
import redis  
  
import threading  
  
import asyncio  
  
# 用户列表  
users = []  
  
# 頻道列表  
channels = ["channel_1","channel_2"]  
  
  
# websocket協議  
class WB(tornado.websocket.WebSocketHandler):  
  
  
    # 跨域支持  
    def check_origin(self,origin):  
  
        return True  
  
    # 開啓鏈接  
    def open(self):  
  
  
        users.append(self)  
  
  
    # 接收消息  
    def on_message(self,message):  
  
        self.write_message(message['data'])  
  
    # 斷開  
    def on_close(self):  
  
        users.remove(self)  
  
  
  
  
  
  
# 基於redis監聽發佈者發佈消息  
def redis_listener(loop):  
  
    asyncio.set_event_loop(loop)  
  
    async def listen():   
  
        r = redis.Redis(decode_responses=True)  
  
        # 聲明pubsb實例  
        ps = r.pubsub()  
  
        # 訂閲聊天室頻道  
  
        ps.subscribe(["channel_1","channel_2"])  
  
  
        # 監聽消息  
        for message in ps.listen():  
  
            print(message)  
  
            # 遍歷鏈接上的用户  
            for user in users:  
  
                print(user)  
  
                if message["type"] == "message" and message["channel"] == user.get_cookie("channel"):  
  
  
                    user.write_message(message["data"])  
  
    future = asyncio.gather(listen())  
    loop.run_until_complete(future)  
  
  
  
# 接口  發佈信息  
class Msg(tornado.web.RequestHandler):  
  
  
    # 重寫父類方法  
    def set_default_headers(self):  
  
        # 設置請求頭信息  
        print("開始設置")  
        # 域名信息  
        self.set_header("Access-Control-Allow-Origin","*")  
        # 請求信息  
        self.set_header("Access-Control-Allow-Headers","x-requested-with")  
        # 請求方式  
        self.set_header("Access-Control-Allow-Methods","POST,GET,PUT,DELETE")  
  
      
  
    # 發佈信息  
    async def post(self):  
  
        data = self.get_argument("data",None)  
  
        channel = self.get_argument("channel","channel_1")  
  
        print(data)  
  
        # 發佈  
        r = redis.Redis()  
  
        r.publish(channel,data)  
  
        return self.write("ok")  
  
  
# 建立torando實例  
  
app = tornado.web.Application(  
  
    [  
  
    (r'/send/',Msg),  
    (r'/wb/',WB)  
  
    ],debug=True  
  
)  
  
if __name__ == '__main__':  
  
  
    loop = asyncio.new_event_loop()  
  
    # 單線程啓動訂閲者服務  
    threading.Thread(target=redis_listener,args=(loop,)).start()  
  
  
    # 聲明服務器  
    http_server_1 = tornado.httpserver.HTTPServer(app)  
  
    # 監聽端口  
    http_server_1.listen(8000)  
  
    # 開啓事件循環  
    tornado.ioloop.IOLoop.instance().start()

這裏假設默認有兩個頻道，邏輯是這樣的：由前端控制websocket鏈接用户選擇將消息發佈到那個頻道上，同時每個用户通過前端cookie的設置具備頻道屬性，當具備頻道屬性的用户對該頻道發佈了一條消息之後，所有其他具備該頻道屬性的用户通過redis進行訂閲後主動推送剛剛發佈的消息，而頻道的推送只匹配訂閲該頻道的用户，達到消息隔離的目的。

需要注意的一點是，通過線程啓動redis訂閲服務時，需要將當前的loop實例傳遞給協程對象，否則在訂閲方法內將會獲取不到websocket實例，報這個錯誤：

IOLoop.current() doesn't work in non-main

這是因為Tornado底層基於事件循環ioloop，而同步框架模式的Django或者Flask則沒有這個問題。

下面編寫前端代碼，這裏我們使用時下最流行的vue3.0框架，編寫chat.vue:

<template>  
  <div>  
  
  
            <h1>聊天窗口</h1>  
  
  
            <van-tabs v-model:active="active" @click="change_channel">  
  
              <van-tab title="客服1號">  
  
  
                <table>  
                
              <tr v-for="item,index in msglist" :key="index">  
                  
                {{ item }}  
  
              </tr>  
  
            </table>  
                  
  
  
              </van-tab>  
  
  
              <van-tab title="客服2號">  
                  
  
                <table>  
                
              <tr v-for="item,index in msglist" :key="index">  
                  
                {{ item }}  
  
              </tr>  
  
            </table>  
  
  
              </van-tab>  
  
            </van-tabs>  
  
  
              
  
  
            <van-field label="聊天信息" v-model="msg" />  
  
            <van-button color="gray" @click="commit">發送</van-button>  
  
     
  </div>  
</template>  
  
<script>  
  
export default {  
 data() {  
    return {  
      auditlist:[],  
  
      //聊天記錄  
      msglist:[],  
      msg:"",  
       websock: null, //建立的連接  
      lockReconnect: false, //是否真正建立連接  
      timeout: 3 * 1000, //30秒一次心跳  
      timeoutObj: null, //外層心跳倒計時  
      serverTimeoutObj: null, //內層心跳檢測  
      timeoutnum: null, //斷開 重連倒計時  
      active:0,  
      channel:"channel_1"  
       
    }  
  },  
  methods:{  
  
  
    //切換頻道  
    change_channel:function(){  
  
  
          if(this.active === 0){  
  
  
                this.channel = "channel_1";  
  
                var name = "channel";  
          var value = "channel_1";  
  
            
  
          }else{  
  
  
              this.channel = "channel_2";  
  
                var name = "channel";  
          var value = "channel_2";  
  
  
          }  
  
  
          //清空聊天記錄  
          this.msglist = [];  
  
  
          var d = new Date();  
          d.setTime(d.getTime() + (24 * 60 * 60 * 1000));  
          var expires = "expires=" + d.toGMTString();  
          document.cookie = name + "=" + value + "; " + expires;  
  
  
          this.reconnect();  
  
  
    },  
     initWebSocket() {  
      //初始化weosocket  
      const wsuri = "ws://localhost:8000/wb/";  
      this.websock = new WebSocket(wsuri);  
      this.websock.onopen = this.websocketonopen;  
      this.websock.onmessage = this.websocketonmessage;  
      this.websock.onerror = this.websocketonerror;  
      this.websock.onclose = this.websocketclose;  
    },  
  
    reconnect() {  
      //重新連接  
      var that = this;  
      if (that.lockReconnect) {  
        // 是否真正建立連接  
        return;  
      }  
      that.lockReconnect = true;  
      //沒連接上會一直重連，設置延遲避免請求過多  
      that.timeoutnum && clearTimeout(that.timeoutnum);  
      // 如果到了這裏斷開重連的倒計時還有值的話就清除掉  
      that.timeoutnum = setTimeout(function() {  
        //然後新連接  
        that.initWebSocket();  
        that.lockReconnect = false;  
      }, 5000);  
    },  
  
     reset() {  
      //重置心跳  
      var that = this;  
      //清除時間（清除內外兩個心跳計時）  
      clearTimeout(that.timeoutObj);  
      clearTimeout(that.serverTimeoutObj);  
      //重啓心跳  
      that.start();  
    },  
  
    start() {  
      //開啓心跳  
      var self = this;  
      self.timeoutObj && clearTimeout(self.timeoutObj);  
      // 如果外層心跳倒計時存在的話，清除掉  
      self.serverTimeoutObj && clearTimeout(self.serverTimeoutObj);  
      // 如果內層心跳檢測倒計時存在的話，清除掉  
      self.timeoutObj = setTimeout(function() {  
        // 重新賦值重新發送 進行心跳檢測  
        //這裏發送一個心跳，後端收到後，返回一個心跳消息，  
        if (self.websock.readyState == 1) {  
          //如果連接正常  
          // self.websock.send("heartCheck");  
        } else {  
          //否則重連  
          self.reconnect();  
        }  
        self.serverTimeoutObj = setTimeout(function() {  
          // 在三秒一次的心跳檢測中如果某個值3秒沒響應就關掉這次連接  
          //超時關閉  
         // self.websock.close();  
        }, self.timeout);  
      }, self.timeout);  
      // 3s一次  
    },  
  
    websocketonopen(e) {  
      //連接建立之後執行send方法發送數據  
      console.log("成功");  
  
     // this.websock.send("123");  
      // this.websocketsend(JSON.stringify(actions));  
    },  
    websocketonerror() {  
      //連接建立失敗重連  
      console.log("失敗");  
      this.initWebSocket();  
    },  
    websocketonmessage(e) {  
  
      console.log(e);  
      //數據接收  
      //const redata = JSON.parse(e.data);  
      const redata = e.data;  
  
      //累加  
      this.msglist.push(redata);  
  
      console.log(redata);  
  
       
    },  
    websocketsend(Data) {  
      //數據發送  
      this.websock.send(Data);  
    },  
    websocketclose(e) {  
      //關閉  
      this.reconnect()  
      console.log("斷開連接", e);  
    },  
  
    //提交表單  
    commit:function(){  
  
  
        //發送請求  
  
        this.myaxios("http://localhost:8000/send/","post",{"data":this.msg,channel:this.channel}).then(data =>{  
  
          console.log(data);  
  
        });  
  
  
  
    },  
    
  
  },  
  
  mounted(){  
  
  
      //連接後端websocket服務  
      this.initWebSocket();  
  
  
  
      var d = new Date();  
          d.setTime(d.getTime() + (24 * 60 * 60 * 1000));  
          var expires = "expires=" + d.toGMTString();  
          document.cookie = "channel" + "=" + "channel_1" + "; " + expires;  
  
      
  
  }  
  
}  
</script>  
  
  
<style scoped>  
  @import url("../assets/style.css");  
  
  .chatbox{  
  
      color:black;  
  
  }  
  
  .mymsg{  
  
      background-color:green;  
  
  }  
  
  
</style>

這裏前端在線客户端定期向狀態服務器發送心跳事件。如果服務端在特定時間內（例如x秒）從客户端接收到心跳事件，則認為用户處於聯機狀態。否則，它將處於脱機狀態，脱機後在閾值時間內可以進行重新連接的動作。同時利用vant框架的標籤頁可以同步切換頻道，切換後將頻道標識寫入cookie，便於後端服務識別後匹配推送。

效果是這樣的：

誠然，功能業已實現，但是如果我們處在一個高併發場景之下呢？試想一下如果一個頻道有10萬人同時在線，每秒有100條新消息，那麼後台tornado的websocket服務推送頻率是100w*10/s = 1000w/s 。

這樣的系統架構如果不做負載均衡的話，很難抗住壓力，那麼瓶頸在哪裏呢？沒錯，就是數據庫redis，這裏我們需要異步redis庫aioredis的幫助：

pip3 install aioredis

aioredis通過協程異步操作redis讀寫，避免了io阻塞問題，使消息的發佈和訂閲操作非阻塞。

此時，可以新建一個異步訂閲服務文件main\_with\_aioredis.py:

import asyncio  
import aioredis  
from tornado import web, websocket  
from tornado.ioloop import IOLoop  
import tornado.httpserver  
import async_timeout

之後主要的修改邏輯是，通過aioredis異步建立redis鏈接，並且異步訂閲多個頻道，隨後通過原生協程的asyncio.create\_task方法(也可以使用asyncio.ensure\_future)註冊訂閲消費的異步任務reader:

async def setup():  
    r = await aioredis.from_url("redis://localhost", decode_responses=True)  
    pubsub = r.pubsub()  
  
    print(pubsub)  
    await pubsub.subscribe("channel_1","channel_2")  
  
    #asyncio.ensure_future(reader(pubsub))  
    asyncio.create_task(reader(pubsub))

在訂閲消費方法中，異步監聽所訂閲頻道中的發佈信息，同時和之前的同步方法一樣，比對用户的頻道屬性並且進行按頻道推送：

async def reader(channel: aioredis.client.PubSub):  
    while True:  
        try:  
            async with async_timeout.timeout(1):  
                message = await channel.get_message(ignore_subscribe_messages=True)  
                if message is not None:  
                    print(f"(Reader) Message Received: {message}")  
  
                    for user in users:  
  
                        if user.get_cookie("channel") == message["channel"]:  
  
                            user.write_message(message["data"])  
          
                await asyncio.sleep(0.01)  
        except asyncio.TimeoutError:  
            pass

最後，利用tornado事件循環IOLoop傳遞中執行回調方法，將setup方法加入到事件回調中：

if __name__ == '__main__':  
  
    # 監聽端口  
    application.listen(8000)  
  
    loop = IOLoop.current()  
    loop.add_callback(setup)  
    loop.start()

完整的異步消息發佈、訂閲、推送服務改造 main\_aioredis.py:

import asyncio  
import aioredis  
from tornado import web, websocket  
from tornado.ioloop import IOLoop  
import tornado.httpserver  
import async_timeout  
  
users = []  
  
# websocket協議  
class WB(tornado.websocket.WebSocketHandler):  
  
  
    # 跨域支持  
    def check_origin(self,origin):  
  
        return True  
  
    # 開啓鏈接  
    def open(self):  
  
  
        users.append(self)  
  
  
    # 接收消息  
    def on_message(self,message):  
  
        self.write_message(message['data'])  
  
    # 斷開  
    def on_close(self):  
  
        users.remove(self)  
  
  
class Msg(web.RequestHandler):  
  
  
    # 重寫父類方法  
    def set_default_headers(self):  
  
        # 設置請求頭信息  
        print("開始設置")  
        # 域名信息  
        self.set_header("Access-Control-Allow-Origin","*")  
        # 請求信息  
        self.set_header("Access-Control-Allow-Headers","x-requested-with")  
        # 請求方式  
        self.set_header("Access-Control-Allow-Methods","POST,GET,PUT,DELETE")  
  
  
    # 發佈信息  
    async def post(self):  
  
        data = self.get_argument("data",None)  
  
        channel = self.get_argument("channel","channel_1")  
  
        print(data)  
  
        # 發佈  
        r = await aioredis.from_url("redis://localhost", decode_responses=True)  
  
        await r.publish(channel,data)  
  
        return self.write("ok")  
  
  
async def reader(channel: aioredis.client.PubSub):  
    while True:  
        try:  
            async with async_timeout.timeout(1):  
                message = await channel.get_message(ignore_subscribe_messages=True)  
                if message is not None:  
                    print(f"(Reader) Message Received: {message}")  
  
                    for user in users:  
  
                        if user.get_cookie("channel") == message["channel"]:  
  
                            user.write_message(message["data"])  
          
                await asyncio.sleep(0.01)  
        except asyncio.TimeoutError:  
            pass  
  
  
async def setup():  
    r = await aioredis.from_url("redis://localhost", decode_responses=True)  
    pubsub = r.pubsub()  
  
    print(pubsub)  
    await pubsub.subscribe("channel_1","channel_2")  
  
    #asyncio.ensure_future(reader(pubsub))  
    asyncio.create_task(reader(pubsub))  
  
  
application = web.Application([  
    (r'/send/',Msg),  
    (r'/wb/', WB),  
],debug=True)      
  
  
if __name__ == '__main__':  
  
    # 監聽端口  
    application.listen(8000)  
  
    loop = IOLoop.current()  
    loop.add_callback(setup)  
    loop.start()

從程序設計角度上講，充分利用了協程的異步執行思想，更加地絲滑流暢。

結語：實踐操作來看，Redis發佈訂閲模式，非常契合這種實時(websocket)通信聊天系統的場景，但是發佈的消息如果沒有對應的頻道或者消費者，消息則會被丟棄，假如我們在生產環境在消費的時候，突然斷網，導致其中一個訂閲者掛掉了一段時間，那麼當它重新連接上的時候，中間這一段時間產生的消息也將不會存在，所以如果想要保證系統的健壯性，還需要其他服務來設計高可用的實時存儲方案，不過那就是另外一個故事了，最後奉上項目地址，與眾鄉親同饗：https://github.com/zcxey2911/...\_redis\_vue3\_chatroom

原文轉載自「劉悦的技術博客」 https://v3u.cn/a_id_202