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

本文涉及的产品
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
云数据库 Tair(兼容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
相关文章
|
2月前
|
并行计算 数据处理 Python
Python并发编程迷雾:IO密集型为何偏爱异步?CPU密集型又该如何应对?
在Python的并发编程世界中,没有万能的解决方案,只有最适合特定场景的方法。希望本文能够为你拨开迷雾,找到那条通往高效并发编程的光明大道。
47 2
|
3月前
|
开发框架 并行计算 算法
揭秘Python并发神器:IO密集型与CPU密集型任务的异步革命,你竟还傻傻分不清?
揭秘Python并发神器:IO密集型与CPU密集型任务的异步革命,你竟还傻傻分不清?
55 4
|
3月前
|
算法 Java 程序员
解锁Python高效之道:并发与异步在IO与CPU密集型任务中的精准打击策略!
在数据驱动时代,高效处理大规模数据和高并发请求至关重要。Python凭借其优雅的语法和强大的库支持,成为开发者首选。本文将介绍Python中的并发与异步编程,涵盖并发与异步的基本概念、IO密集型任务的并发策略、CPU密集型任务的并发策略以及异步IO的应用。通过具体示例,展示如何使用`concurrent.futures`、`asyncio`和`multiprocessing`等库提升程序性能,帮助开发者构建高效、可扩展的应用程序。
157 0
|
4月前
|
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
|
5月前
|
并行计算 数据处理 Python
Python并发编程迷雾:IO密集型为何偏爱异步?CPU密集型又该如何应对?
【7月更文挑战第17天】Python并发编程中,异步编程(如`asyncio`)在IO密集型任务中提高效率,利用等待时间执行其他任务。但对CPU密集型任务,由于GIL限制,多线程效率不高,此时应选用`multiprocessing`进行多进程并行计算以突破限制。选择合适的并发策略是关键:异步适合IO,多进程适合CPU。理解这些能帮助构建高效并发程序。
129 6
|
5月前
|
算法 Java 程序员
解锁Python高效之道:并发与异步在IO与CPU密集型任务中的精准打击策略!
【7月更文挑战第17天】在数据驱动时代,Python凭借其优雅语法和强大库支持成为并发处理大规模数据的首选。并发与异步编程是关键,包括多线程、多进程和异步IO。对于IO密集型任务,如网络请求,可使用`concurrent.futures`和`asyncio`;CPU密集型任务则推荐多进程,如`multiprocessing`;`asyncio`适用于混合任务,实现等待IO时执行CPU任务。通过这些工具,开发者能有效优化资源,提升系统性能。
104 4
|
5月前
|
开发框架 并行计算 .NET
从菜鸟到大神:Python并发编程深度剖析,IO与CPU的异步战争!
【7月更文挑战第18天】Python并发涉及多线程、多进程和异步IO(asyncio)。异步IO适合IO密集型任务,如并发HTTP请求,能避免等待提高效率。多进程在CPU密集型任务中更优,因可绕过GIL限制实现并行计算。通过正确选择并发策略,开发者能提升应用性能和响应速度。
119 3
|
5月前
|
开发框架 并行计算 算法
揭秘Python并发神器:IO密集型与CPU密集型任务的异步革命,你竟还傻傻分不清?
【7月更文挑战第18天】Python并发编程中,异步IO适合IO密集型任务,如异步HTTP请求,利用`asyncio`和`aiohttp`实现并发抓取,避免等待延迟。而对于CPU密集型任务,如并行计算斐波那契数列,多进程通过`multiprocessing`库能绕过GIL限制实现并行计算。选择正确的并发模型能显著提升性能。
96 2
|
5月前
|
开发框架 数据挖掘 .NET
显微镜下的Python并发:细说IO与CPU密集型任务的异步差异,助你精准施策!
【7月更文挑战第16天】在Python并发编程中,理解和区分IO密集型与CPU密集型任务至关重要。IO密集型任务(如网络请求)适合使用异步编程(如`asyncio`),以利用等待时间执行其他任务,提高效率。CPU密集型任务(如计算)则推荐使用多进程(如`multiprocessing`),绕过GIL限制,利用多核CPU。正确选择并发策略能优化应用性能。
84 2
|
5月前
|
SQL 关系型数据库 MySQL
实时计算 Flink版产品使用问题之在Flink算子内部使用异步IO可以通过什么办法实现
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。