Using Flask and Celery to push real-time/timed messages asynchronously in Win10 environment (Socket.io)/The latest strategy in 2020

    首先得明确一点,和Django一样,在2020年Flask 1.1.1Later versions do not require so-called third-party library support,即Flask-Celery或者Flask-Celery-Help这些库,直接使用CeleryThe native library can be used.

    一般情况下,CeleryUsed to handle time-consuming tasks,Such as stereotyped emails or file uploads,本次使用CeleryReal-time or scheduled delivery basedWebsocket的消息队列,Because if the front end has abandoned the old polling strategy,使用Websocket,The back end needs to be matched accordinglyCeleryMake pairs persistentWebsocketLinks to proactively push messages,This scenario is still very common in production environments,But there are few articles on the Internet,而CeleryThe official statement on this is：

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

    Basically it means：因为 Celery 和 前端Web 是分开的 Process So there needs to be a common backend to trigger the push of messages,This is an availabilityCelery触发WebsocketThe focus of message push.

    第一步,安装必须的库

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

    flask-corsLibraries are libraries used to circumvent the browser's same-origin policy,flask-socketioUsed to establish full duplexwebsocket链接,celeryTakes responsibility for an asynchronous task queue.

    实例化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'

    The message queue container is still used hereredis

    Then use the initializedapp队列,初始化socket对象,This will allow based onwsgi的Flask支持websocket

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

    这里注意下,Plus cross-domain parameters,And specify the asynchronous mode as thread.

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

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

    After that, you can declare some necessary methods and views,and run the instance

@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)

    You can see the asynchronous call task [email protected]()来声明,而基于websocketview [email protected]来声明,在Flask项目的目录下,分别开启两个命令行,启动Web服务和Celery服务

python manage.py

    启动celery服务

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

    这里celeryThe service is still based on the coroutine libraryeventlet

    The front end uses the more popular ones on the marketVue.js,需要安装socket.io的支持

npm install [email protected]

    Write a component for testingclient.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:[]
    }
  },
  //Register component tags
  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>

    By listening on the same key as the backend“sendback”to display the messages pushed in the background.

    Test out asynchronous push

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

    The front end immediately receives the messages asynchronously pushed by the back end.

    Now let's test the scheduled task,基于Celery的CrontabThe advantage is that it supports second-level timing,在上面celery初始化之后,You can define scheduled tasks through configuration

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


celery.conf.CELERYBEAT_SCHEDULE = {
        
        "test":{
            "task":"get_cron",
            "schedule":timedelta(seconds=10)
        }

}

    Here we add a test task,定时每10Push a message in seconds

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

    Just asynchronously call the push method you just wrote,This allows you to share a backend with the frontendwebsocket链接,Otherwise, the scheduled task cannot trigger the message push.

    Start a third service in the same directory,注意webServices and asynchronous services do not stop

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

    You can see the timing pushwebsocketThe message also came true.

    This feature is essentially an application-level decoupling,用Celery特有的taskway to basewebsocket推送emit消息,二者相辅相成.

    Finally serve thisdemo的版本库：https://gitee.com/QiHanXiBei/myflask