需求：本次專案需求是自己用flask設計一個api，接受一些引數，然後使用佇列動態將引數傳給後臺的worker，在此總結一下redis的pub# -*- coding: utf-8 -*-from flask import Flask

from flask import request, jsonify
from redis import Redis

app = Flask(__name__)


@app.route("/v1/enc-aab", methods=["POST"])
def first_post():
    """函式主體"""
    data = request.get_json()
    path 
 
= data.get('path')
    name = data.get('name')
    res = {'path': path, 'name': name}
    print(res)
    # 業務處理
    if not all([path, name]):
        return jsonify(msg='引數不完整')
    else:
        # 資料完整 開始pub
        rc = Redis(host='127.0.0.1', port=6379, db=0)
        rc.pubsub()
        rc.publish('
 
res', res)
        return jsonify(status='ok')


if __name__ == '__main__':
    app.run()

到此為止就在介面中拿到了傳遞的data並pub進了queue

pub的總結如下：

# ##### 
rc = Redis(host='127.0.0.1', port=6379, db=0)
rc.pubsub()
rc.publish('res', res)
# #####          

sub的應用如下：

# ###### 
rc = Redis(host='127.0.0.1', port=6379, db=0)
ps 
 
= rc.pubsub()
ps.subscribe(['res'])
for item in ps.listen():
   print(item)
# ######