把酒言欢话聊天，基于Vue3.0+Tornado6.1+Redis发布订阅(pubsub)模式打造异步非阻塞(aioredis)实时(websocket)通信聊天系统

原文转载自「刘悦的技术博客」https://v3u.cn/a_id_202

“表达欲”是人类成长史上的强大“源动力”&＃xff0c;恩格斯早就直截了当地指出&＃xff0c;处在蒙昧时代即低级阶段的人类&＃xff0c;“以果实、坚果、根作为食物&＃xff1b;音节清晰的语言的产生是这一时期的主要成就”。而在网络时代人们的表达欲往往更容易被满足&＃xff0c;因为有聊天软件的存在。通常意义上&＃xff0c;聊天大抵都基于两种形式&＃xff1a;群聊和单聊。群聊或者群组聊天我们可以理解为聊天室&＃xff0c;可以有人数上限&＃xff0c;而单聊则可以认为是上限为2个人的特殊聊天室。

为了开发高质量的聊天系统&＃xff0c;开发者应该具备客户机和服务器如何通信的基本知识。在聊天系统中&＃xff0c;客户端可以是移动应用程序(C端)或web应用程序(B端)。客户端之间不直接通信。相反&＃xff0c;每个客户端都连接到一个聊天服务&＃xff0c;该服务支撑双方通信的功能。所以该服务在业务上必须支持的最基本功能&＃xff1a;

1.能够实时接收来自其他客户端的信息。

2.能够将每条信息实时推送给收件人。

当客户端打算启动聊天时&＃xff0c;它会使用一个或多个网络协议连接聊天服务。对于聊天服务&＃xff0c;网络协议的选择至关重要&＃xff0c;这里&＃xff0c;我们选择Tornado框架内置Websocket协议的接口&＃xff0c;简单而又方便&＃xff0c;安装tornado6.1

pip3 install tornado&＃61;&＃61;6.1


随后编写程序启动文件main.py:

import tornado.httpserver
import tornado.websocket import tornado.ioloop import tornado.web import redis import threading import asyncio # 用户列表
users &＃61; [] # 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[&＃39;data&＃39;]) # 断开 def on_close(self): users.remove(self)# 建立torando实例 app &＃61; tornado.web.Application( [ (r&＃39;/wb/&＃39;,WB) ],debug&＃61;True ) if __name__ &＃61;&＃61; &＃39;__main__&＃39;: # 声明服务器 http_server_1 &＃61; tornado.httpserver.HTTPServer(app) # 监听端口 http_server_1.listen(8000) # 开启事件循环 tornado.ioloop.IOLoop.instance().start()


如此&＃xff0c;就在短时间搭建起了一套websocket协议服务&＃xff0c;每一次有客户端发起websocket连接请求&＃xff0c;我们都会将它添加到用户列表中&＃xff0c;等待用户的推送或者接收信息的动作。

下面我们需要通过某种形式将消息的发送方和接收方联系起来&＃xff0c;以达到“聊天”的目的&＃xff0c;这里选择Redis的发布订阅模式(pubsub)&＃xff0c;以一个demo来实例说明&＃xff0c;server.py

import redis r &＃61; redis.Redis()
r.publish("test",&＃39;hello&＃39;)


随后编写 client.py:

import redis
r &＃61; redis.Redis()
ps &＃61; r.pubsub()
ps.subscribe(&＃39;test&＃39;)
for item in ps.listen(): if item[&＃39;type&＃39;] &＃61;&＃61; &＃39;message&＃39;: print(item[&＃39;data&＃39;])


可以这么理解&＃xff1a;订阅者(listener)负责订阅频道(channel)&＃xff1b;发送者(publisher)负责向频道(channel)发送二进制的字符串消息&＃xff0c;然后频道收到消息时&＃xff0c;推送给订阅者。

频道不仅可以联系发布者和订阅者&＃xff0c;同时&＃xff0c;也可以利用频道进行“消息隔离”&＃xff0c;即不同频道的消息只会给订阅该频道的用户进行推送&＃xff1a;

根据发布者订阅者逻辑&＃xff0c;改写main.py:

import tornado.httpserver
import tornado.websocket import tornado.ioloop import tornado.web import redis import threading import asyncio # 用户列表
users &＃61; [] # 频道列表
channels &＃61; ["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[&＃39;data&＃39;]) # 断开 def on_close(self): users.remove(self) # 基于redis监听发布者发布消息
def redis_listener(loop): asyncio.set_event_loop(loop) async def listen(): r &＃61; redis.Redis(decode_responses&＃61;True) # 声明pubsb实例 ps &＃61; r.pubsub() # 订阅聊天室频道 ps.subscribe(["channel_1","channel_2"]) # 监听消息 for message in ps.listen(): print(message) # 遍历链接上的用户 for user in users: print(user) if message["type"] &＃61;&＃61; "message" and message["channel"] &＃61;&＃61; user.get_COOKIE("channel"): user.write_message(message["data"]) future &＃61; 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 &＃61; self.get_argument("data",None) channel &＃61; self.get_argument("channel","channel_1") print(data) # 发布 r &＃61; redis.Redis() r.publish(channel,data) return self.write("ok") # 建立torando实例 app &＃61; tornado.web.Application( [ (r&＃39;/send/&＃39;,Msg), (r&＃39;/wb/&＃39;,WB) ],debug&＃61;True ) if __name__ &＃61;&＃61; &＃39;__main__&＃39;: loop &＃61; asyncio.new_event_loop() # 单线程启动订阅者服务 threading.Thread(target&＃61;redis_listener,args&＃61;(loop,)).start() # 声明服务器 http_server_1 &＃61; tornado.httpserver.HTTPServer(app) # 监听端口 http_server_1.listen(8000) # 开启事件循环 tornado.ioloop.IOLoop.instance().start()


这里假设默认有两个频道&＃xff0c;逻辑是这样的&＃xff1a;由前端控制websocket链接用户选择将消息发布到那个频道上&＃xff0c;同时每个用户通过前端COOKIE的设置具备频道属性&＃xff0c;当具备频道属性的用户对该频道发布了一条消息之后&＃xff0c;所有其他具备该频道属性的用户通过redis进行订阅后主动推送刚刚发布的消息&＃xff0c;而频道的推送只匹配订阅该频道的用户&＃xff0c;达到消息隔离的目的。

需要注意的一点是&＃xff0c;通过线程启动redis订阅服务时&＃xff0c;需要将当前的loop实例传递给协程对象&＃xff0c;否则在订阅方法内将会获取不到websocket实例&＃xff0c;报这个错误&＃xff1a;

IOLoop.current() doesn&＃39;t work in non-main


这是因为Tornado底层基于事件循环ioloop&＃xff0c;而同步框架模式的Django或者Flask则没有这个问题。

下面编写前端代码&＃xff0c;这里我们使用时下最流行的vue3.0框架&＃xff0c;编写chat.vue:

这里前端在线客户端定期向状态服务器发送心跳事件。如果服务端在特定时间内&＃xff08;例如x秒&＃xff09;从客户端接收到心跳事件&＃xff0c;则认为用户处于联机状态。否则&＃xff0c;它将处于脱机状态&＃xff0c;脱机后在阈值时间内可以进行重新连接的动作。同时利用vant框架的标签页可以同步切换频道&＃xff0c;切换后将频道标识写入COOKIE&＃xff0c;便于后端服务识别后匹配推送。

效果是这样的&＃xff1a;

诚然&＃xff0c;功能业已实现&＃xff0c;但是如果我们处在一个高并发场景之下呢&＃xff1f;试想一下如果一个频道有10万人同时在线&＃xff0c;每秒有100条新消息&＃xff0c;那么后台tornado的websocket服务推送频率是100w*10/s &＃61; 1000w/s 。

这样的系统架构如果不做负载均衡的话&＃xff0c;很难抗住压力&＃xff0c;那么瓶颈在哪里呢&＃xff1f;没错&＃xff0c;就是数据库redis&＃xff0c;这里我们需要异步redis库aioredis的帮助&＃xff1a;

pip3 install aioredis


aioredis通过协程异步操作redis读写&＃xff0c;避免了io阻塞问题&＃xff0c;使消息的发布和订阅操作非阻塞。

此时&＃xff0c;可以新建一个异步订阅服务文件main_with_aioredis.py:

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


之后主要的修改逻辑是&＃xff0c;通过aioredis异步建立redis链接&＃xff0c;并且异步订阅多个频道&＃xff0c;随后通过原生协程的asyncio.create_task方法(也可以使用asyncio.ensure_future)注册订阅消费的异步任务reader:

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


在订阅消费方法中&＃xff0c;异步监听所订阅频道中的发布信息&＃xff0c;同时和之前的同步方法一样&＃xff0c;比对用户的频道属性并且进行按频道推送&＃xff1a;

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


最后&＃xff0c;利用tornado事件循环IOLoop传递中执行回调方法&＃xff0c;将setup方法加入到事件回调中&＃xff1a;

if __name__ &＃61;&＃61; &＃39;__main__&＃39;: # 监听端口 application.listen(8000) loop &＃61; 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 &＃61; [] # 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[&＃39;data&＃39;]) # 断开 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 &＃61; self.get_argument("data",None) channel &＃61; self.get_argument("channel","channel_1") print(data) # 发布 r &＃61; await aioredis.from_url("redis://localhost", decode_responses&＃61;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 &＃61; await channel.get_message(ignore_subscribe_messages&＃61;True) if message is not None: print(f"(Reader) Message Received: {message}") for user in users: if user.get_COOKIE("channel") &＃61;&＃61; message["channel"]: user.write_message(message["data"]) await asyncio.sleep(0.01) except asyncio.TimeoutError: pass async def setup(): r &＃61; await aioredis.from_url("redis://localhost", decode_responses&＃61;True) pubsub &＃61; r.pubsub() print(pubsub) await pubsub.subscribe("channel_1","channel_2") #asyncio.ensure_future(reader(pubsub)) asyncio.create_task(reader(pubsub)) application &＃61; web.Application([ (r&＃39;/send/&＃39;,Msg), (r&＃39;/wb/&＃39;, WB),
],debug&＃61;True) if __name__ &＃61;&＃61; &＃39;__main__&＃39;: # 监听端口 application.listen(8000) loop &＃61; IOLoop.current() loop.add_callback(setup) loop.start()


从程序设计角度上讲&＃xff0c;充分利用了协程的异步执行思想&＃xff0c;更加地丝滑流畅。

结语&＃xff1a;实践操作来看&＃xff0c;Redis发布订阅模式&＃xff0c;非常契合这种实时(websocket)通信聊天系统的场景&＃xff0c;但是发布的消息如果没有对应的频道或者消费者&＃xff0c;消息则会被丢弃&＃xff0c;假如我们在生产环境在消费的时候&＃xff0c;突然断网&＃xff0c;导致其中一个订阅者挂掉了一段时间&＃xff0c;那么当它重新连接上的时候&＃xff0c;中间这一段时间产生的消息也将不会存在&＃xff0c;所以如果想要保证系统的健壮性&＃xff0c;还需要其他服务来设计高可用的实时存储方案&＃xff0c;不过那就是另外一个故事了&＃xff0c;最后奉上项目地址&＃xff0c;与众乡亲同飨&＃xff1a;https://github.com/zcxey2911/tornado_redis_vue3_chatroom

原文转载自「刘悦的技术博客」 https://v3u.cn/a_id_202