怎麼用redis+python做訊息佇列

一、使用redis的List型別結合lpush 和brpop 來實現

簡介

• #首先redis的List 相當於一個佇列，可以實現先進先出的規則

• 採用brpop 是因為當佇列中沒有的時候會被阻塞，直到佇列中有可彈出元素或等待逾時

##模擬問題：

• 訪問太多，伺服器處理速度太慢，如果每隔用戶等待，伺服器回饋的話，時間太長，http連線逾時，出現伺服器錯誤。

模擬實作過程：

• 有一個客戶端不斷的往佇列裡放東西(資料)，採用多線程，模擬大量用戶訪問的情況

• 有一個伺服器不斷的從佇列中取出列印，並設定每次列印時間睡2秒

redis的List結構介紹

key [value, value]
key 代表List的名字， [value, ...] 是值

客戶client.py

import random
import threading
import redis
import config

lock = threading.Lock()
lock.acquire()
lock.release()


pool = redis.ConnectionPool(host=config.HOST, port=config.PORT, 
                            decode_responses=True, password=config.PASSWORD)

r = redis.Redis(connection_pool=pool)

# 客户往redis 中放数据
def fun1(redisObj):
    value = random.randint(0, 100)
    # 往ccc列表中存放
    print("开始发送数据:", value)
    redisObj.lpush("print",str(value))

for i in range(100):
    threading.Thread(target=fun1, args=(r,)).start()

伺服器server.py

import redis
import time
import config


pool = redis.ConnectionPool(host=config.HOST, port=config.PORT, decode_responses=True, password=config.PASSWORD)
r = redis.Redis(connection_pool=pool)
# 服务端不断的取
while True:
    value = r.brpop("print")
    time.sleep(2)
    print(value)

問題回顧

我們之前說存在阻塞太久斷開連線的問題，解決下

方式： 將連接作為一個函數，進行錯誤捕捉，發生問題的時候重新連接。

import redis
import time
import config

def get_redis():
    pool = redis.ConnectionPool(host=config.HOST, port=config.PORT, decode_responses=True, password=config.PASSWORD)
    r = redis.Redis(connection_pool=pool)
    return r
# 服务端不断的取
r = get_redis()
    
while True:
    try:
        value = r.brpop("print")
        time.sleep(2)
        print(value)
    except Exception as e:
        print("等待超时重连")
        r = get_redis()

以上是怎麼用redis+python做訊息佇列的詳細內容。更多資訊請關注PHP中文網其他相關文章！