Die Flask-Framework-Anwendung von Python ruft Redis-Warteschlangendaten auf

Aufgabe asynchron

Öffnen Sie den Browser, geben Sie die Adresse ein, drücken Sie die Eingabetaste und öffnen Sie die Seite. Daher wird eine HTTP-Anfrage (Anfrage) vom Client an den Server gesendet, und der Server verarbeitet die Anfrage und gibt den Inhalt der Antwort (Antwort) zurück.

Wir surfen jeden Tag im Internet und senden große und kleine Anfragen an den Server. Wenn der Server eine Anfrage empfängt, stellt er manchmal fest, dass er auch eine Anfrage an einen anderen Server senden muss, oder der Server muss auch etwas anderes tun, sodass die ursprünglich gesendete Anfrage blockiert wird, das heißt, sie muss warten damit der Server andere Dinge erledigen kann.

Häufiger erfordern die vom Server ausgeführten zusätzlichen Aufgaben keine Wartezeit des Clients. Zu diesem Zeitpunkt können diese zusätzlichen Aufgaben asynchron ausgeführt werden. Es gibt viele Tools zum Ausführen asynchroner Aufgaben. Das Hauptprinzip besteht immer noch darin, Benachrichtigungsnachrichten zu verarbeiten. Für Benachrichtigungsnachrichten wird normalerweise eine Warteschlangenstruktur verwendet. Produzieren und konsumieren Sie Nachrichten für die Kommunikation und Geschäftsumsetzung.

Produktion, Verbrauch und Warteschlange
Die Implementierung der oben genannten asynchronen Aufgaben kann in das Produzentenverbrauchsmodell abstrahiert werden. Genau wie in einem Restaurant kocht der Koch und die Feinschmecker essen. Wenn der Koch viel macht und vorerst nicht alle Artikel verkaufen kann, macht der Koch eine Pause, wenn es viele Kunden gibt und der Koch ununterbrochen beschäftigt ist, müssen die Kunden langsam warten. Es gibt viele Möglichkeiten, Produzenten und Konsumenten zu implementieren. Hier ist ein kleines Beispiel mit der Python-Standardbibliothek 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()

Die ungefähre Ausgabe ist wie folgt:

厨师 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-Warteschlange
Python verfügt über eine integrierte nützliche Warteschlangenstruktur. Wir können Redis auch verwenden, um ähnliche Vorgänge zu implementieren. und eine einfache asynchrone Aufgabe ausführen.

Redis bietet zwei Möglichkeiten, Nachrichtenwarteschlangen zu erstellen. Zum einen wird das Producer-Consumer-Modell verwendet, zum anderen das Publish-Subscriber-Modell. Ersteres ermöglicht es einem oder mehreren Clients, die Nachrichtenwarteschlange zu überwachen. Sobald die Nachricht eintrifft, ist derjenige, der sie zuerst erhält, der Gewinner. Wenn sich keine Nachricht in der Warteschlange befindet, hört der Verbraucher weiter zu. Letzteres bedeutet auch, dass ein oder mehrere Clients den Nachrichtenkanal abonnieren. Solange der Herausgeber die Nachricht veröffentlicht, können alle Abonnenten die Nachricht empfangen und die Abonnenten werden angepingt.

Produktions- und Verbrauchsmodus
Verwendet hauptsächlich von Redis bereitgestelltes Blpop, um Warteschlangendaten abzurufen. Wenn sich keine Daten in der Warteschlange befinden, wird es blockiert und gewartet, dh abgehört.

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

Veröffentlichungs- und Abonnementmodell
Mithilfe der Pubsub-Funktion von Redis abonniert der Abonnent den Kanal und den Herausgeber veröffentlicht Nachrichten im Kanal Ja, ein Kanal ist eine Nachrichtenwarteschlange.

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

Flask-Eingang
Wir haben die Backend-Dienste von jeweils zwei asynchronen Aufgaben implementiert und sie jetzt direkt gestartet Überwachen Sie Nachrichten aus Redis-Warteschlangen oder -Kanälen. Ein einfacher Test sieht wie folgt aus:

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)

Um das Skript zu starten, verwenden Sie

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

Asynchrone Nachrichten können separat in der abgehörten Skripteingabe angezeigt werden. Bei asynchronen Aufgaben können Sie einige zeitaufwändige Vorgänge ausführen. Natürlich kennen diese Methoden derzeit die asynchronen Ausführungsergebnisse nicht. Wenn Sie die asynchronen Ausführungsergebnisse kennen müssen, können Sie erwägen, Coroutine-Aufgaben zu entwerfen oder einige Tools wie RQ zu verwenden oder Sellerie.

Weitere Artikel zu Pythons Flask-Framework-Anwendung zum Aufrufen von Redis-Warteschlangendaten finden Sie auf der chinesischen PHP-Website!