基于django-channel将redis发布的消息实时推送到前端

蔚落 2022-12-23 07:11 305阅读 0赞

# 0 Redis发布与订阅模式 #

订阅频道channel\_01

127.0.0.1:6379> SUBSCRIBE channel_01

向指定频道发布

127.0.0.1:6379> PUBLISH channel_01 "info1"

发布消息后，订阅的命令行窗口会展示新的消息

127.0.0.1:6379> SUBSCRIBE channel_01
    Reading messages... (press Ctrl-C to quit)
    1) "subscribe"
    2) "channel_01"
    3) (integer) 1
    1) "message"
    2) "channel_01"
    3) "info1"

# 1 Python实现Redis发布订阅模式 #

监听频道：

import redis
    
    
    if __name__ == '__main__':
        rc = redis.StrictRedis(host='127.0.0.1', port='6379', password='password')
        ps = rc.pubsub()
        ps.subscribe('info_01')
        for item in ps.listen():
            if item['type'] == 'message':
                channel = item['channel'].decode()
                message = item['data'].decode()
                print(channel, "=>", message)

发布消息：

import redis
    
    if __name__ == '__main__':
        rc = redis.StrictRedis(host='127.0.0.1', port='6379', db=0, password='password')
        rc.publish("info_01", '{"id": 3, "count": 10}')
        rc.publish("info_01", '{"id": 4, "count": 16}')
        rc.publish("info_01", '{"id": 5, "count": 20}')

# 2 django另起线程监听redis频道发布 #

在django中使用redis，这里提供两种方式：

*  一种是直接使用python redis包，在项目启动或者需要的时候，创建redis连接对象。
 *  第二种方式是使用django-redis包，在settings中配置redis连接消息，可以随时调用连接池和redis连接，不需要我们手动编写代码管理redis连接对象。  
    这里第二种为例，先安装`django-redis`：`pip install django-redis`

安装好后，在settings中进行配置：

CACHES = {
        
        "default": {
        
            "BACKEND": "django_redis.cache.RedisCache",
            "LOCATION": "redis://127.0.0.1:6379/1",
            "OPTIONS": {
        
                "CONNECTION_POOL_KWARGS": {
        "max_connections": 20},
                "PASSWORD": "password"
            }
        }
    }

然后可以使用`django-redis`提供的原生连接对象（get\_redis\_connection）、连接池（connection\_pool）、包装后的对象（cache）等。  
这里使用连接对象，因为连接池和包装后的对象暂不支持订阅和发布操作。  
在views.py中编写另起线程的代码：

import _thread
    from django.core.cache import cache
    from django_redis import get_redis_connection
    
    # 启动线程
    def start_information_subscribe():
        try:
            print("Start: 尝试启动线程")
            _thread.start_new_thread(push_service, ())
        except Exception:
            print("Error: 无法启动线程")
    
    # 线程执行的方法
    def push_service():
    	# 订阅频道
        con = get_redis_connection("default")
        ps = con.pubsub()
        ps.subscribe('abcdefg')
    
        def push(channel, msg):
            push2somewhere(channel, msg)
    	
    	# 监听发布
        for item in ps.listen():
            if item['type'] == 'message':
                channel = item['channel'].decode()
                message = item['data'].decode()
                push(channel, message)
    
    
    # django服务启动时也启动该线程
    # 为了实现线程高可用，上生产应使用celery+supervisor
    start_information_subscribe()

这样，就可以在django服务内拥有一个监听redis消息队列的线程了，在频道内接收到消息时，我们可以调用push2somewhere方法把消息存到其他数据、推送到前端、执行某个视图等。

# 3 将redis发布的消息实时推送到前端 #

当我们监听redis频道的线程监听到消息之后，如何推送到前端呢？  
毕竟，http不支持主动像客户端推送。因此，我们这里使用websocket来实现此功能。  
django构建websocket应用可以参考：[基于django-channel构建websocket的web应用、支持token校验][django-channel_websocket_web_token]  
看了上面的文章，我们知道向websocket客户端推送消息，可以调用group\_send向指定的channel发送消息。  
因此，我们把上面的push(channel, msg)方法改成如下：

from asgiref.sync import async_to_sync
    from channels.layers import get_channel_layer
    def push_service():
    	......
    	def push(channel, msg):
    	    channel_layer = get_channel_layer()
    	    async_to_sync(channel_layer.group_send)(
    	        "WS_{}".format(channel),
    	        {
        "type": "push_message", "message": msg}
    	    )
    	......

这样，websocket客户端就可以实时收到redis推送的消息了。

# 4 创建websocket客户端进行验证 #

这里不提供前端代码的实现，提供使用asynio创建websocket客户端的代码：

import asyncio
    import json
    
    import websockets
    
    # 向服务器端发送认证后的消息
    async def send_msg(websocket):
        while True:
            recv_text = await websocket.recv()
            print(recv_text)
    
    
    # 客户端主逻辑
    async def main_logic():
        async with websockets.connect('ws://127.0.0.1:8000/ws/base/report_id/',
                                      extra_headers={
        "Authorization": "Token 174077a7f"}
                                      ) as websocket:
            await send_msg(websocket)
    
    asyncio.get_event_loop().run_until_complete(main_logic())

[django-channel_websocket_web_token]: https://blog.csdn.net/weixin_39198406/article/details/110006822