L'application-cadre Flask de Python appelle les données de la file d'attente Redis

Tâche asynchrone

Ouvrez le navigateur, saisissez l'adresse, appuyez sur Entrée et ouvrez la page. Ainsi, une requête HTTP (requête) est envoyée du client au serveur, et le serveur traite la requête et renvoie le contenu de la réponse (réponse).

Nous parcourons le Web tous les jours et envoyons des requêtes grandes et petites au serveur. Parfois, lorsque le serveur reçoit une requête, il constatera qu'il doit également envoyer une requête à un autre serveur, ou que le serveur doit également faire autre chose, donc la requête qu'il a initialement envoyée est bloquée, c'est-à-dire qu'il doit attendre pour que le serveur effectue d'autres tâches.

Le plus souvent, les tâches supplémentaires effectuées par le serveur ne nécessitent pas que le client attende. À ce stade, ces tâches supplémentaires peuvent être effectuées de manière asynchrone. Il existe de nombreux outils pour effectuer des tâches asynchrones. Le principe principal reste de traiter les messages de notification. Pour les messages de notification, une structure de file d'attente est généralement adoptée. Produire et consommer des messages pour la communication et la mise en œuvre commerciale.

Production, consommation et file d'attente
La mise en œuvre des tâches asynchrones ci-dessus peut être abstraite dans le modèle de consommation du producteur. Tout comme dans un restaurant, le chef cuisine et les gourmands mangent. Si le chef gagne beaucoup et ne peut pas vendre tous les articles pour le moment, le chef fera une pause ; s'il y a beaucoup de clients et que le chef est occupé sans arrêt, les clients doivent attendre lentement. Il existe de nombreuses façons d'implémenter des producteurs et des consommateurs. Voici un petit exemple utilisant la bibliothèque standard Python Queue :

import random
import time
from Queue import Queue
from threading import Thread

queue = Queue(10)

class Producer(Thread):
  def run(self):
    while True:
      elem = random.randrange(9)
      queue.put(elem)
      print "厨师 {} 做了 {} 饭 --- 还剩 {} 饭没卖完".format(self.name, elem, queue.qsize())
      time.sleep(random.random())

class Consumer(Thread):
  def run(self):
    while True:
      elem = queue.get()
      print "吃货{} 吃了 {} 饭 --- 还有 {} 饭可以吃".format(self.name, elem, queue.qsize())
      time.sleep(random.random())

def main():
  for i in range(3):
    p = Producer()
    p.start()
  for i in range(2):
    c = Consumer()
    c.start()

if __name__ == '__main__':
  main()

Le résultat approximatif est le suivant :

厨师 Thread-1 做了 1 饭 --- 还剩 1 饭没卖完
厨师 Thread-2 做了 8 饭 --- 还剩 2 饭没卖完
厨师 Thread-3 做了 3 饭 --- 还剩 3 饭没卖完
吃货Thread-4 吃了 1 饭 --- 还有 2 饭可以吃
吃货Thread-5 吃了 8 饭 --- 还有 1 饭可以吃
吃货Thread-4 吃了 3 饭 --- 还有 0 饭可以吃
厨师 Thread-1 做了 0 饭 --- 还剩 1 饭没卖完
厨师 Thread-2 做了 0 饭 --- 还剩 2 饭没卖完
厨师 Thread-1 做了 1 饭 --- 还剩 3 饭没卖完
厨师 Thread-1 做了 1 饭 --- 还剩 4 饭没卖完
吃货Thread-4 吃了 0 饭 --- 还有 3 饭可以吃
厨师 Thread-3 做了 3 饭 --- 还剩 4 饭没卖完
吃货Thread-5 吃了 0 饭 --- 还有 3 饭可以吃
吃货Thread-5 吃了 1 饭 --- 还有 2 饭可以吃
厨师 Thread-2 做了 8 饭 --- 还剩 3 饭没卖完
厨师 Thread-2 做了 8 饭 --- 还剩 4 饭没卖完

Redis Queue
Python a une structure de file d'attente utile intégrée. Nous pouvons également utiliser Redis pour implémenter des opérations similaires. et effectuez une simple tâche asynchrone.

Redis propose deux façons de créer des files d'attente de messages. L’une consiste à utiliser le modèle de consommation du producteur et l’autre est le modèle de publication-abonnement. Le premier permettra à un ou plusieurs clients de surveiller la file d'attente des messages. Une fois le message arrivé, le consommateur le consommera immédiatement. Celui qui le récupérera en premier sera le gagnant. S'il n'y a pas de message dans la file d'attente, le consommateur continuera à écouter. Ce dernier est également un ou plusieurs clients abonnés au canal de message. Tant que l'éditeur publie le message, tous les abonnés peuvent recevoir le message et les abonnés sont pingés.

Mode de production et de consommation
Utilise principalement le blpop fourni par redis pour obtenir les données de la file d'attente, s'il n'y a pas de données dans la file d'attente, il bloquera et attendra, c'est-à-dire écoutera.

import redis

class Task(object):
  def __init__(self):
    self.rcon = redis.StrictRedis(host='localhost', db=5)
    self.queue = 'task:prodcons:queue'

  def listen_task(self):
    while True:
      task = self.rcon.blpop(self.queue, 0)[1]
      print "Task get", task

if __name__ == '__main__':
  print 'listen task queue'
  Task().listen_task()

Modèle de publication et d'abonnement
À l'aide de la fonction pubsub de redis, l'abonné s'abonne à la chaîne et à l'éditeur publie des messages sur le canal Oui, un canal est une file d'attente de messages.

import redis


class Task(object):

  def __init__(self):
    self.rcon = redis.StrictRedis(host='localhost', db=5)
    self.ps = self.rcon.pubsub()
    self.ps.subscribe('task:pubsub:channel')

  def listen_task(self):
    for i in self.ps.listen():
      if i['type'] == 'message':
        print "Task get", i['data']

if __name__ == '__main__':
  print 'listen task channel'
  Task().listen_task()

Entrée Flask
Nous avons implémenté les services backend de deux tâches asynchrones respectivement et les avons démarrées directement Can maintenant. surveiller les messages des files d'attente ou des canaux Redis. Un test simple est le suivant :

import redis
import random
import logging
from flask import Flask, redirect

app = Flask(__name__)

rcon = redis.StrictRedis(host='localhost', db=5)
prodcons_queue = 'task:prodcons:queue'
pubsub_channel = 'task:pubsub:channel'

@app.route('/')
def index():

  html = """
<br>
<center><h3>Redis Message Queue</h3>
<br>
<a href="/prodcons">生产消费者模式</a>
<br>
<br>
<a href="/pubsub">发布订阅者模式</a>
</center>
"""
  return html


@app.route(&#39;/prodcons&#39;)
def prodcons():
  elem = random.randrange(10)
  rcon.lpush(prodcons_queue, elem)
  logging.info("lpush {} -- {}".format(prodcons_queue, elem))
  return redirect(&#39;/&#39;)

@app.route(&#39;/pubsub&#39;)
def pubsub():
  ps = rcon.pubsub()
  ps.subscribe(pubsub_channel)
  elem = random.randrange(10)
  rcon.publish(pubsub_channel, elem)
  return redirect(&#39;/&#39;)

if __name__ == &#39;__main__&#39;:
  app.run(debug=True)

pour démarrer le script, utilisez

siege -c10 -r 5 http://127.0.0.1:5000/prodcons
siege -c10 -r 5 http://127.0.0.1:5000/pubsub

Les messages asynchrones peuvent être vus séparément dans l'entrée du script écouté. Dans les tâches asynchrones, vous pouvez effectuer certaines opérations fastidieuses. Bien entendu, ces méthodes actuelles ne connaissent pas les résultats de l'exécution asynchrone, vous pouvez envisager de concevoir des tâches coroutines ou d'utiliser certains outils tels que RQ. ou du céleri.

Pour plus d'articles liés à l'application-cadre Flask de Python appelant les données de la file d'attente Redis, veuillez faire attention au site Web PHP chinois !