个人中心
gpt4 book ai didi

Python的Flask框架应用调用Redis队列数据的方法

转载 作者：qq735679552 更新时间：2022-09-29 22:32:09 26 4
gpt4 key购买 nike

CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章Python的Flask框架应用调用Redis队列数据的方法由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.

任务异步化 打开浏览器，输入地址，按下回车，打开了页面。于是一个HTTP请求（request）就由客户端发送到服务器，服务器处理请求，返回响应（response）内容.

我们每天都在浏览网页，发送大大小小的请求给服务器。有时候，服务器接到了请求，会发现他也需要给另外的服务器发送请求，或者服务器也需要做另外一些事情，于是最初们发送的请求就被阻塞了，也就是要等待服务器完成其他的事情.

更多的时候，服务器做的额外事情，并不需要客户端等待，这时候就可以把这些额外的事情异步去做。从事异步任务的工具有很多。主要原理还是处理通知消息，针对通知消息通常采取是队列结构。生产和消费消息进行通信和业务实现.

生产消费与队列 上述异步任务的实现，可以抽象为生产者消费模型。如同一个餐馆，厨师在做饭，吃货在吃饭。如果厨师做了很多，暂时卖不完，厨师就会休息；如果客户很多，厨师马不停蹄的忙碌，客户则需要慢慢等待。实现生产者和消费者的方式用很多，下面使用Python标准库Queue写个小例子:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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()

大概输出如下:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
厨师 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提供了两种方式来作消息队列。一个是使用生产者消费模式模式，另外一个方法就是发布订阅者模式。前者会让一个或者多个客户端监听消息队列，一旦消息到达，消费者马上消费，谁先抢到算谁的，如果队列里没有消息，则消费者继续监听。后者也是一个或多个客户端订阅消息频道，只要发布者发布消息，所有订阅者都能收到消息，订阅者都是ping的.

生产消费模式 主要使用了redis提供的blpop获取队列数据，如果队列没有数据则阻塞等待，也就是监听.

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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功能，订阅者订阅频道，发布者发布消息到频道了，频道就是一个消息队列.

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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队列或频道的消息了。简单的测试如下:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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( '/prodcons' )
def prodcons():
   elem = random.randrange( 10 )
   rcon.lpush(prodcons_queue, elem)
   logging.info( "lpush {} -- {}" . format (prodcons_queue, elem))
   return redirect( '/' )
 
@app .route( '/pubsub' )
def pubsub():
   ps = rcon.pubsub()
   ps.subscribe(pubsub_channel)
   elem = random.randrange( 10 )
   rcon.publish(pubsub_channel, elem)
   return redirect( '/' )
 
if __name__ = = '__main__' :
   app.run(debug = True )

启动脚本，使用 。

?
1
2
siege -c10 -r 5 http: //127 .0.0.1:5000 /prodcons
siege -c10 -r 5 http: //127 .0.0.1:5000 /pubsub

可以分别在监听的脚本输入中看到异步消息。在异步的任务中，可以执行一些耗时间的操作，当然目前这些做法并不知道异步的执行结果，如果需要知道异步的执行结果，可以考虑设计协程任务或者使用一些工具如RQ或者celery等.

最后此篇关于Python的Flask框架应用调用Redis队列数据的方法的文章就讲到这里了,如果你想了解更多关于Python的Flask框架应用调用Redis队列数据的方法的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。

26 4 0
文章推荐： Xcode 9下适配iPhoneX导致iOS 10不兼容问题的解决方法
文章推荐： iOS实现带指引线的饼状图效果（不会重叠）
文章推荐： Python第三方库的安装方法总结
文章推荐： ios wkwebview离线化加载h5资源解决方案
qq735679552
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
全站热门文章
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com