Redis實現簡單訊息佇列

阿新 • • 發佈：2019-01-13

任務非同步化

開啟瀏覽器，輸入地址，按下回車，打開了頁面。於是一個HTTP請求（request）就由客戶端傳送到伺服器，伺服器處理請求，返回響應（response）內容。

我們每天都在瀏覽網頁，傳送大大小小的請求給伺服器。有時候，伺服器接到了請求，會發現他也需要給另外的伺服器傳送請求，或者伺服器也需要做另外一些事情，於是最初們傳送的請求就被阻塞了，也就是要等待伺服器完成其他的事情。

更多的時候，伺服器做的額外事情，並不需要客戶端等待，這時候就可以把這些額外的事情非同步去做。從事非同步任務的工具有很多。主要原理還是處理通知訊息，針對通知訊息通常採取是佇列結構。生產和消費訊息進行通訊和業務實現。

生產消費與佇列

上述非同步任務的實現，可以抽象為生產者消費模型。如同一個餐館，廚師在做飯，吃貨在吃飯。如果廚師做了很多，暫時賣不完，廚師就會休息；如果客戶很多，廚師馬不停蹄的忙碌，客戶則需要慢慢等待。實現生產者和消費者的方式用很多，下面使用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()

1234567891011121314151617181920212223242526272829303132

import randomimport timefrom Queue import Queuefrom threading import Threadqueue=Queue(10)classProducer(Thread):def run(self):whileTrue:elem=random.randrange(9)queue.put(elem)print"廚師 {} 做了 {} 飯 --- 還剩 {} 飯沒賣完"

.format(self.name,elem,queue.qsize())time.sleep(random.random())classConsumer(Thread):def run(self):whileTrue:elem=queue.get()print"吃貨{} 吃了 {} 飯 --- 還有 {} 飯可以吃".format(self.name,elem,queue.qsize())time.sleep(random.random())def main():foriinrange(3):p=Producer()p.start()foriinrange(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 飯沒賣完

12345678910111213141516

廚師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獲取佇列資料，如果佇列沒有資料則阻塞等待，也就是監聽。

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

123456789101112131415

import redisclassTask(object):def __init__(self):self.rcon=redis.StrictRedis(host='localhost',db=5)self.queue='task:prodcons:queue'def listen_task(self):whileTrue:task=self.rcon.blpop(self.queue,0)[1]print"Task get",taskif__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()

1234567891011121314151617

import redisclassTask(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):foriinself.ps.listen():ifi['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('/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)

12345678910111213141516171819202122232425262728293031323334353637383940414243

import redisimport randomimport loggingfrom flask import Flask,redirectapp=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>"""returnhtml@app.route('/prodcons')def prodcons():elem=random.randrange(10)rcon.lpush(prodcons_queue,elem)logging.info("lpush {} -- {}".format(prodcons_queue,elem))returnredirect('/')@app.route('/pubsub')def pubsub():ps=rcon.pubsub()ps.subscribe(pubsub_channel)elem=random.randrange(10)rcon.publish(pubsub_channel,elem)returnredirect('/')if__name__=='__main__':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

12	siege-c10-r5http://127.0.0.1:5000/prodconssiege-c10-r5http://127.0.0.1:5000/pubsub

可以分別在監聽的指令碼輸入中看到非同步訊息。在非同步的任務中，可以執行一些耗時間的操作，當然目前這些做法並不知道非同步的執行結果，如果需要知道非同步的執行結果，可以考慮設計協程任務或者使用一些工具如RQ或者celery等。

Redis實現簡單訊息佇列

任務非同步化

生產消費與佇列

Redis 佇列

生產消費模式

釋出訂閱模式

Flask 入口

Redis實現簡單訊息佇列

利用Redis實現非同步訊息佇列優化系統性能（Redis高階應用）

基於Redis實現分散式訊息佇列（彙總目錄）

基於Redis實現分散式訊息佇列（1）

Spring Spring-data-redis 實現的訊息佇列

使用Redis實現非同步訊息佇列

基於Redis實現分散式訊息佇列（4）

基於Redis實現分散式訊息佇列（3）

什麼鬼，面試官竟然讓我用Redis實現一個訊息佇列！！？

用 Redis 實現 PHP 的簡單訊息佇列

Redis延時訊息佇列、非同步訊息佇列的實現

玩轉redis-簡單訊息佇列

進階高階IoT架構-教你如何簡單實現一個訊息佇列

使用redis實現簡單的秒殺

Redis-實現簡單的條件查詢

Delayer 基於 Redis 的延遲訊息佇列中介軟體

spring-boot+Redis實現簡單的分散式叢集session共享

php實現的訊息佇列類

Netty實戰開發(7):Netty結合kafka實現分散式訊息佇列

java 連結redis 實現簡單的查詢

Redis實現簡單訊息佇列

任務非同步化

生產消費與佇列

Redis 佇列

生產消費模式

釋出訂閱模式

Flask 入口

相關推薦