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

相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
相关文章
|
1月前
|
并行计算 数据处理 Python
Python并发编程迷雾:IO密集型为何偏爱异步?CPU密集型又该如何应对?
在Python的并发编程世界中,没有万能的解决方案,只有最适合特定场景的方法。希望本文能够为你拨开迷雾,找到那条通往高效并发编程的光明大道。
42 2
|
2月前
|
开发框架 并行计算 算法
揭秘Python并发神器:IO密集型与CPU密集型任务的异步革命,你竟还傻傻分不清?
揭秘Python并发神器:IO密集型与CPU密集型任务的异步革命,你竟还傻傻分不清?
42 4
|
3月前
|
NoSQL Redis UED
揭秘!Flask如何携手Celery,让异步任务处理不再是难题,打造极速响应的Web应用新纪元!
【8月更文挑战第31天】在Web开发中,Flask与Celery的结合为异步任务处理提供了强大支持。Flask作为轻量级Web框架,以其简洁灵活著称;而Celery则是一个分布式任务队列系统,擅长处理耗时任务。二者结合,Flask专注于处理Web请求,Celery则在后台异步执行复杂任务,如发送邮件或调用外部API。这种方式不仅提升了应用性能和响应速度,还优化了用户体验。下面的示例展示了如何在Flask应用中集成Celery以实现异步任务处理。
136 0
|
2月前
|
NoSQL 大数据 Redis
使用 Flask 和 Celery 构建异步任务处理
使用 Flask 和 Celery 构建异步任务处理
|
2月前
|
数据采集 NoSQL 调度
flask celery python 每月定时任务
flask celery python 每月定时任务
|
2月前
|
算法 Java 程序员
解锁Python高效之道:并发与异步在IO与CPU密集型任务中的精准打击策略!
在数据驱动时代,高效处理大规模数据和高并发请求至关重要。Python凭借其优雅的语法和强大的库支持,成为开发者首选。本文将介绍Python中的并发与异步编程,涵盖并发与异步的基本概念、IO密集型任务的并发策略、CPU密集型任务的并发策略以及异步IO的应用。通过具体示例,展示如何使用`concurrent.futures`、`asyncio`和`multiprocessing`等库提升程序性能,帮助开发者构建高效、可扩展的应用程序。
104 0
|
3月前
|
数据库 关系型数据库 MySQL
惊!Hibernate与MySQL的绝密优化技巧大揭秘,让你的数据库飞起来!
【8月更文挑战第31天】在企业应用开发中,结合使用持久层框架Hibernate与数据库管理系统MySQL可显著提升数据库交互效率。本文探讨了多项优化策略,包括配置二级缓存、采用单向关联减少JOIN操作、优化HQL查询语句以及合理使用MySQL索引。通过具体示例,文章详细讲解了如何实施这些优化措施,以期为企业应用提供更高效稳定的数据支持。
58 0
|
3月前
|
Linux Python
【Azure 应用服务】Azure App Service For Linux 上实现 Python Flask Web Socket 项目 Http/Https
【Azure 应用服务】Azure App Service For Linux 上实现 Python Flask Web Socket 项目 Http/Https
|
4月前
|
并行计算 数据处理 Python
Python并发编程迷雾:IO密集型为何偏爱异步?CPU密集型又该如何应对?
【7月更文挑战第17天】Python并发编程中,异步编程(如`asyncio`)在IO密集型任务中提高效率,利用等待时间执行其他任务。但对CPU密集型任务,由于GIL限制,多线程效率不高,此时应选用`multiprocessing`进行多进程并行计算以突破限制。选择合适的并发策略是关键:异步适合IO,多进程适合CPU。理解这些能帮助构建高效并发程序。
113 6
|
4月前
|
算法 Java 程序员
解锁Python高效之道:并发与异步在IO与CPU密集型任务中的精准打击策略!
【7月更文挑战第17天】在数据驱动时代,Python凭借其优雅语法和强大库支持成为并发处理大规模数据的首选。并发与异步编程是关键,包括多线程、多进程和异步IO。对于IO密集型任务,如网络请求,可使用`concurrent.futures`和`asyncio`;CPU密集型任务则推荐多进程,如`multiprocessing`;`asyncio`适用于混合任务,实现等待IO时执行CPU任务。通过这些工具,开发者能有效优化资源,提升系统性能。
90 4