Python의 Flask 프레임워크 애플리케이션은 Redis 대기열 데이터를 호출합니다.

작업 비동기

브라우저를 열고 주소를 입력한 후 Enter 키를 누르고 페이지를 엽니다. 그래서 클라이언트에서 서버로 HTTP 요청(요청)이 전송되고, 서버는 요청을 처리하고 응답(응답) 내용을 반환합니다.

우리는 매일 웹을 탐색하며 서버에 크고 작은 요청을 보냅니다. 때로는 서버가 요청을 받으면 다른 서버에도 요청을 보내야 하거나 서버도 다른 작업을 수행해야 한다는 사실을 알게 되어 원래 보낸 요청이 차단됩니다. 즉, 기다려야 합니다. 서버가 다른 작업을 완료하도록 합니다.

서버에서 수행되는 추가 작업은 클라이언트가 기다릴 필요가 없는 경우가 더 많습니다. 이때 이러한 추가 작업은 비동기식으로 수행될 수 있습니다. 비동기 작업을 수행하기 위한 많은 도구가 있습니다. 주요 원칙은 여전히 ​​알림 메시지를 처리하는 것입니다. 알림 메시지의 경우 일반적으로 대기열 구조가 채택됩니다. 통신 및 비즈니스 구현을 위한 메시지를 생성하고 소비합니다.

생산, 소비 및 대기열
위 비동기 작업의 구현은 생산자 소비 모델로 추상화될 수 있습니다. 레스토랑과 마찬가지로 셰프가 요리를 하고 미식가들이 식사를 합니다. 요리사가 요리를 많이 해서 당분간 모든 음식을 팔 수 없다면 잠시 쉬어야 하고, 손님이 많아 쉬지 않고 바쁘다면 손님은 천천히 기다려야 합니다. 생산자와 소비자를 구현하는 방법에는 여러 가지가 있습니다. 다음은 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()

대략적인 출력은 다음과 같습니다.

厨师 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 대기열
Python에는 유용한 대기열 구조가 내장되어 있습니다. Redis를 사용하여 유사한 작업을 구현할 수도 있습니다. 간단한 비동기 작업을 수행합니다.

Redis는 메시지 대기열을 수행하는 두 가지 방법을 제공합니다. 하나는 생산자 소비 모델을 사용하는 것이고, 다른 하나는 게시-구독자 모델을 사용하는 것입니다. 전자는 하나 이상의 클라이언트가 메시지 대기열을 모니터링하도록 하며, 메시지가 도착하면 소비자가 즉시 이를 소비합니다. 대기열에 메시지가 없으면 소비자는 계속해서 메시지를 듣게 됩니다. 후자는 게시자가 메시지를 게시하는 한 메시지 채널을 구독하는 하나 이상의 클라이언트이기도 합니다. 모든 구독자는 메시지를 받을 수 있으며 구독자에게 핑이 전송됩니다.

생산 및 소비 모드
큐 데이터를 얻기 위해 주로 redis에서 제공하는 blpop을 사용합니다. 큐에 데이터가 없으면 차단하고 대기합니다. 즉, 대기합니다.

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

게시 및 구독 모델
redis의 pubsub 기능을 사용하여 구독자는 채널을 구독하고 게시자는 게시합니다. 채널에 대한 메시지는 메시지 대기열입니다.

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 입구
우리는 비동기 작업을 위해 두 가지 종류의 백엔드 서비스를 구현했습니다. Redis 큐 또는 채널의 메시지를 모니터링할 수 있습니다. 간단한 테스트는 다음과 같습니다:

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)

스크립트를 시작하려면

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

스크립트를 각각 시작하려면 리스너의 스크립트 입력에 비동기 메시지가 표시됩니다. 비동기 작업에서는 시간이 많이 걸리는 작업을 수행할 수 있습니다. 물론 이러한 메서드는 현재 비동기 실행 결과를 알 수 없습니다. 비동기 실행 결과를 알아야 하는 경우 코루틴 작업을 설계하거나 RQ와 같은 일부 도구를 사용하는 것을 고려할 수 있습니다. 아니면 셀러리.

Redis 큐 데이터를 호출하는 Python의 Flask 프레임워크 애플리케이션과 관련된 더 많은 기사를 보려면 PHP 중국어 웹사이트를 주목하세요!