Win11环境下使用Flask配合Celery异步推送实时/定时消息(Socket.io)

本文涉及的产品
云数据库 Tair(兼容Redis),内存型 2GB
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
简介: 一般情况下,Celery被用来处理耗时任务,比如千篇一律的发邮件或者文件上传之类,本次使用Celery实时或者定时发送基于Websocket的消息队列,因为如果前端已经摒弃老旧的轮询策略,使用Websocket,后端则需要相应的配合Celery进行对持久化的Websocket链接主动推送消息,这种场景在生产环境中还是很常见的,但是网上却鲜有文章阐述,而Celery官方对此的说明是

首先得明确一点,和Django一样,在2020年Flask 1.1.1以后的版本都不需要所谓的三方库支持,即Flask-Celery或者Flask-Celery-Help这些库,直接使用Celery原生库即可。

一般情况下,Celery被用来处理耗时任务,比如千篇一律的发邮件或者文件上传之类,本次使用Celery实时或者定时发送基于Websocket的消息队列,因为如果前端已经摒弃老旧的轮询策略,使用Websocket,后端则需要相应的配合Celery进行对持久化的Websocket链接主动推送消息,这种场景在生产环境中还是很常见的,但是网上却鲜有文章阐述,而Celery官方对此的说明是:

If using multiple processes, a message queue service is used by the processes to coordinate operations such as broadcasting. The supported queues are Redis, RabbitMQ, and any other message queues supported by the Kombu package

大体上的意思是:因为 Celery 和 前端Web 是分开的 Process 所以需要有一个共同的后端来触发消息的推送,这是一个能否用Celery触发Websocket消息推送的重点。

第一步,安装必须的库

pip3 install flask-cors  
pip3 install flask-socketio  
pip3 install celery

flask-cors库是用来规避浏览器同源策略的库,flask-socketio用来建立全双工websocket链接,celery承担异步任务队列的职责。

实例化app对象

from flask_cors import CORS  
from flask_socketio import SocketIO,send,emit,join_room, leave_room  
import urllib.parse  
from celery import Celery  
from datetime import timedelta  
   
app = Flask(__name__)  
  
app.config['BROKER_URL'] = 'redis://localhost:6379'  
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379'  
app.config['CELERY_ACCEPT_CONTENT'] = ['json', 'pickle']  
app.config['REDIS_URL'] = 'redis://localhost:6379'

这里消息队列容器还是使用redis

随后利用初始化的app队列,初始化socket对象,这样才能让基于wsgi的Flask支持websocket

socketio = SocketIO(app,cors_allowed_origins='*',async_mode="threading",message_queue=app.config['CELERY_RESULT_BACKEND'])

这里注意下,加上跨域参数,并且指定异步模式为线程。

第三步,就是初始化celery对象

celery = Celery(app.name)  
celery.conf.update(app.config)

之后就可以声明一些必要的方法和视图,并且运行实例

@celery.task()  
def get_sendback():  
      
    socketio.emit('sendback','message',broadcast=True)  
  
@app.route('/task')  
def start_background_task():  
    get_sendback.delay()  
    return '开始'  
   
  
@socketio.on('join')  
def on_join(data):  
    username = 'user1'  
    room = 'room1'  
    join_room(room)  
    send(username + ' has entered the room.', room=room)  
  
@socketio.on('message')  
def handle_message(message):  
    message = urllib.parse.unquote(message)  
    print(message)  
    send(message,broadcast=True)  
  
@socketio.on('connect', namespace='/chat')  
def test_connect():  
    emit('my response', {'data': 'Connected'})  
  
@socketio.on('disconnect', namespace='/chat')  
def test_disconnect():  
    print('Client disconnected')  
   
@app.route("/sendback",methods=['GET'])  
def sendback():  
  
    socketio.emit('sendback','message')  
  
    return 'ok'  
   
if __name__ == '__main__':  
    app.config['JSON_AS_ASCII'] = False  
    socketio.run(app,debug=True,host="0.0.0.0",port=5000)

可以看到异步调用任务使用@celery.task()来声明,而基于websocket的视图则用@socketio.on来声明,在Flask项目的目录下,分别开启两个命令行,启动Web服务和Celery服务

python manage.py

启动celery服务

celery worker -A manage.celery --loglevel=info -P eventlet

这里celery服务还是基于协程库eventlet

前端使用市面上比较流行的Vue.js,需要安装socket.io的支持

npm install vue-socket.io@2.1.0

编写一个用来测试的组件client.vue

<template>  
  <div>  
  
    <div v-for="item in log_list"  
    >  
    {{item}}  
  </div>  
  
    <input v-model="msg" />  
  
    <button @click="send">发送消息</button>  
  
  
</div>  
    
</template>  
  
  
   
<script>  
  
export default {  
  data () {  
    return {  
      msg: "",  
      log_list:[]  
    }  
  },  
  //注册组件标签  
  components:{  
  
  
  },  
  sockets:{  
    connect: function(){  
      console.log('socket 连接成功')  
    },  
    message: function(val){  
      console.log('返回:'+val);  
    alert(val);  
      this.log_list.push(val);  
    },  
sendback: function(val){  
      console.log('返回:'+val);  
    alert(val);  
    }  
},  
  mounted:function(){  
  
        
},  
  methods:{  
  
    send(){  
    this.$socket.emit('join',encodeURI("加入房间"))  
      this.$socket.emit('message',encodeURI("用户:"+this.msg));  
  
    },  
    
  
       
  }  
}  
  
  
</script>  
   
<style>  
  
  
  
</style>

通过监听和后端相同的键“sendback”来展示后台推送的消息。

测试一下异步推送

访问url触发异步任务:http://localhost:5000/sendback

前端立刻受到了后端异步推送的消息。

下面我们来测试一下定时任务,基于Celery的Crontab好处就是支持秒级定时,在上面celery初始化之后,就可以通过配置的方式定义定时任务

celery = Celery(app.name)  
celery.conf.update(app.config)  
  
  
celery.conf.CELERYBEAT_SCHEDULE = {  
          
        "test":{  
            "task":"get_cron",  
            "schedule":timedelta(seconds=10)  
        }  
  
}

这里我们增加一个测试任务,定时每10秒推送一条消息

@celery.task(name="get_cron")  
def get_cron():  
      
    get_sendback.delay()

直接异步调用刚刚写好的推送方法即可,这样就可以和前端共用一个后端websocket链接,否则定时任务就无法触发消息推送。

同一目录下启动第三个服务,注意web服务和异步服务不要停

celery -A manage.celery beat --loglevel=debug

可以看到定时推送websocket消息也实现了。

这个功能本质上就是一个应用层面的解耦,用Celery特有的task方式来基于websocket推送emit消息,二者相辅相成。

最后奉上这个demo的版本库:https://gitee.com/QiHanXiBei/myflask

相关文章
|
并行计算 数据处理 Python
Python并发编程迷雾:IO密集型为何偏爱异步?CPU密集型又该如何应对?
在Python的并发编程世界中,没有万能的解决方案,只有最适合特定场景的方法。希望本文能够为你拨开迷雾,找到那条通往高效并发编程的光明大道。
182 2
|
开发框架 并行计算 算法
揭秘Python并发神器:IO密集型与CPU密集型任务的异步革命,你竟还傻傻分不清?
揭秘Python并发神器:IO密集型与CPU密集型任务的异步革命,你竟还傻傻分不清?
158 4
|
NoSQL Redis UED
揭秘!Flask如何携手Celery,让异步任务处理不再是难题,打造极速响应的Web应用新纪元!
【8月更文挑战第31天】在Web开发中,Flask与Celery的结合为异步任务处理提供了强大支持。Flask作为轻量级Web框架,以其简洁灵活著称;而Celery则是一个分布式任务队列系统,擅长处理耗时任务。二者结合,Flask专注于处理Web请求,Celery则在后台异步执行复杂任务,如发送邮件或调用外部API。这种方式不仅提升了应用性能和响应速度,还优化了用户体验。下面的示例展示了如何在Flask应用中集成Celery以实现异步任务处理。
600 0
|
NoSQL 大数据 Redis
使用 Flask 和 Celery 构建异步任务处理
使用 Flask 和 Celery 构建异步任务处理
600 2
|
数据采集 NoSQL 调度
flask celery python 每月定时任务
flask celery python 每月定时任务
|
算法 Java 程序员
解锁Python高效之道:并发与异步在IO与CPU密集型任务中的精准打击策略!
在数据驱动时代,高效处理大规模数据和高并发请求至关重要。Python凭借其优雅的语法和强大的库支持,成为开发者首选。本文将介绍Python中的并发与异步编程,涵盖并发与异步的基本概念、IO密集型任务的并发策略、CPU密集型任务的并发策略以及异步IO的应用。通过具体示例,展示如何使用`concurrent.futures`、`asyncio`和`multiprocessing`等库提升程序性能,帮助开发者构建高效、可扩展的应用程序。
508 0
|
并行计算 数据处理 Python
Python并发编程迷雾:IO密集型为何偏爱异步?CPU密集型又该如何应对?
【7月更文挑战第17天】Python并发编程中,异步编程(如`asyncio`)在IO密集型任务中提高效率,利用等待时间执行其他任务。但对CPU密集型任务,由于GIL限制,多线程效率不高,此时应选用`multiprocessing`进行多进程并行计算以突破限制。选择合适的并发策略是关键:异步适合IO,多进程适合CPU。理解这些能帮助构建高效并发程序。
282 6
|
JSON 数据格式 Python
Flask实现内部接口----pycharm安装及新建,location代表着文件路径,下面是Python的环境,Flask是由Python开发的框架,Python文件接口ython通过GET发送
Flask实现内部接口----pycharm安装及新建,location代表着文件路径,下面是Python的环境,Flask是由Python开发的框架,Python文件接口ython通过GET发送
|
算法 Java 程序员
解锁Python高效之道:并发与异步在IO与CPU密集型任务中的精准打击策略!
【7月更文挑战第17天】在数据驱动时代,Python凭借其优雅语法和强大库支持成为并发处理大规模数据的首选。并发与异步编程是关键,包括多线程、多进程和异步IO。对于IO密集型任务,如网络请求,可使用`concurrent.futures`和`asyncio`;CPU密集型任务则推荐多进程,如`multiprocessing`;`asyncio`适用于混合任务,实现等待IO时执行CPU任务。通过这些工具,开发者能有效优化资源,提升系统性能。
246 4
|
开发框架 并行计算 .NET
从菜鸟到大神:Python并发编程深度剖析,IO与CPU的异步战争!
【7月更文挑战第18天】Python并发涉及多线程、多进程和异步IO(asyncio)。异步IO适合IO密集型任务,如并发HTTP请求,能避免等待提高效率。多进程在CPU密集型任务中更优,因可绕过GIL限制实现并行计算。通过正确选择并发策略,开发者能提升应用性能和响应速度。
251 3