使用redis 進(jìn)行簡(jiǎn)單的隊(duì)列很容易，不需要使用較為復(fù)雜的MQ隊(duì)列，直接使用redis 進(jìn)行，不過(guò)唯一不足的需要自己構(gòu)造生產(chǎn)者消費(fèi)者，這里使用while True的方法進(jìn)行消費(fèi)者操作

介紹

key-value 存儲(chǔ)系統(tǒng)，是跨平臺(tái)的非關(guān)系型數(shù)據(jù)庫(kù)。Redis 通常被稱為數(shù)據(jù)結(jié)構(gòu)服務(wù)器，因?yàn)橹?#xff08;value）可以是字符串(String)、哈希(Hash)、列表(list)、集合(sets)和有序集合(sorted sets)等類型。

常用于緩存、消息隊(duì)列、會(huì)話存儲(chǔ)等應(yīng)用場(chǎng)景。

• **性能極高：**Redis 以其極高的性能而著稱，能夠支持每秒數(shù)十萬(wàn)次的讀寫(xiě)操作24。這使得Redis成為處理高并發(fā)請(qǐng)求的理想選擇，尤其是在需要快速響應(yīng)的場(chǎng)景中，如緩存、會(huì)話管理、排行榜等。
• **豐富的數(shù)據(jù)類型：**Redis 不僅支持基本的鍵值存儲(chǔ)，還提供了豐富的數(shù)據(jù)類型，包括字符串、列表、集合、哈希表、有序集合等。這些數(shù)據(jù)類型為開(kāi)發(fā)者提供了靈活的數(shù)據(jù)操作能力，使得Redis可以適應(yīng)各種不同的應(yīng)用場(chǎng)景。
• **原子性操作：**Redis 的所有操作都是原子性的，這意味著操作要么完全執(zhí)行，要么完全不執(zhí)行。這種特性對(duì)于確保數(shù)據(jù)的一致性和完整性至關(guān)重要，尤其是在高并發(fā)環(huán)境下處理事務(wù)時(shí)。
• **持久化：**Redis 支持?jǐn)?shù)據(jù)的持久化，可以將內(nèi)存中的數(shù)據(jù)保存到磁盤中，以便在系統(tǒng)重啟后恢復(fù)數(shù)據(jù)。這為 Redis 提供了數(shù)據(jù)安全性，確保數(shù)據(jù)不會(huì)因?yàn)橄到y(tǒng)故障而丟失。
• **支持發(fā)布/訂閱模式：**Redis 內(nèi)置了發(fā)布/訂閱模式（Pub/Sub），允許客戶端之間通過(guò)消息傳遞進(jìn)行通信。這使得 Redis 可以作為消息隊(duì)列和實(shí)時(shí)數(shù)據(jù)傳輸?shù)钠脚_(tái)。
• **單線程模型：**盡管 Redis 是單線程的，但它通過(guò)高效的事件驅(qū)動(dòng)模型來(lái)處理并發(fā)請(qǐng)求，確保了高性能和低延遲。單線程模型也簡(jiǎn)化了并發(fā)控制的復(fù)雜性。
• **主從復(fù)制：**Redis 支持主從復(fù)制，可以通過(guò)從節(jié)點(diǎn)來(lái)備份數(shù)據(jù)或分擔(dān)讀請(qǐng)求，提高數(shù)據(jù)的可用性和系統(tǒng)的伸縮性。
• **應(yīng)用場(chǎng)景廣泛：**Redis 被廣泛應(yīng)用于各種場(chǎng)景，包括但不限于緩存系統(tǒng)、會(huì)話存儲(chǔ)、排行榜、實(shí)時(shí)分析、地理空間數(shù)據(jù)索引等。
• **社區(qū)支持：**Redis 擁有一個(gè)活躍的開(kāi)發(fā)者社區(qū)，提供了大量的文檔、教程和第三方庫(kù)，這為開(kāi)發(fā)者提供了強(qiáng)大的支持和豐富的資源。
• **跨平臺(tái)兼容性：**Redis 可以在多種操作系統(tǒng)上運(yùn)行，包括 Linux、macOS 和 Windows，這使得它能夠在不同的技術(shù)棧中靈活部署。
  ? **支持 Lua 腳本：**Redis 支持使用 Lua 腳本來(lái)編寫(xiě)復(fù)雜的操作，這些腳本可以在服務(wù)器端執(zhí)行，提供了更多的靈活性和強(qiáng)大的功能。

數(shù)據(jù)類型

Redis主要支持以下幾種數(shù)據(jù)類型：

• string（字符串）:

  基本的數(shù)據(jù)存儲(chǔ)單元，可以存儲(chǔ)字符串、整數(shù)或者浮點(diǎn)數(shù)。

• hash（哈希）：

  一個(gè)鍵值對(duì)集合，可以存儲(chǔ)多個(gè)字段。

• list（列表）：

  一個(gè)簡(jiǎn)單的列表，可以存儲(chǔ)一系列的字符串元素。

• set（集合）：

  一個(gè)無(wú)序集合，可以存儲(chǔ)不重復(fù)的字符串元素。

• zset(sorted set：社群集合)：

  相似集合，但是每個(gè)元素都有一個(gè)分?jǐn)?shù)（score）關(guān)聯(lián)。

• 位圖（Bitmaps）：

  基于操作字符串類型，可以對(duì)每個(gè)位進(jìn)行。

• 超日志（HyperLogLogs）：

  用于基本統(tǒng)計(jì)，可以提示集合中的唯一元素?cái)?shù)量。

• 地理空間（Geospatial）：

  用于存儲(chǔ)斷層信息。

• 發(fā)布/訂閱（Pub/Sub）：

  一種消息通信模式，允許客戶端訂閱消息通道，并接收發(fā)布到該通道的消息。

• 流（Streams）：

  用于消息隊(duì)列和日志存儲(chǔ)，支持消息的持久化和時(shí)間排序。

• 模塊（Modules）：

  Redis支持動(dòng)態(tài)加載模塊，可以擴(kuò)展Redis的功能。

String

string 是 redis 最基本的類型,string類型最大存儲(chǔ)512MB

• SET key value：設(shè)置鍵的值。
• GET key：獲取鍵的值。
• INCR key：將鍵的值加 1。
• DECR key：將鍵的值減 1。
• APPEND key value：將值追加到鍵的值之后。

> set run "1"

Hash

Redis hash 是一個(gè)鍵值(key=>value)對(duì)集合，類似于一個(gè)小型的 NoSQL 數(shù)據(jù)庫(kù)。

Redis hash 是一個(gè) string 類型的 field 和 value 的映射表，hash 特別適合用于存儲(chǔ)對(duì)象。

• HSET key field value：設(shè)置哈希表中字段的值。
• HGET key field：獲取哈希表中字段的值。
• HGETALL key：獲取哈希表中所有字段和值。
• HDEL key field：刪除哈希表中的一個(gè)或多個(gè)字段。

最常用的還是string，其他用到了再說(shuō)。

Redis支持多個(gè)數(shù)據(jù)庫(kù)，并且每個(gè)數(shù)據(jù)庫(kù)的數(shù)據(jù)是隔離的不能共享，并且基于單機(jī)才有，如果是集群就沒(méi)有數(shù)據(jù)庫(kù)的概念。

Redis是一個(gè)字典結(jié)構(gòu)的存儲(chǔ)服務(wù)器，而實(shí)際上一個(gè)Redis實(shí)例提供了多個(gè)用來(lái)存儲(chǔ)數(shù)據(jù)的字典，客戶端可以指定將數(shù)據(jù)存儲(chǔ)在哪個(gè)字典中。這與我們熟知的在一個(gè)關(guān)系數(shù)據(jù)庫(kù)實(shí)例中可以創(chuàng)建多個(gè)數(shù)據(jù)庫(kù)類似，所以可以將其中的每個(gè)字典都理解成一個(gè)獨(dú)立的數(shù)據(jù)庫(kù)。

每個(gè)數(shù)據(jù)庫(kù)對(duì)外都是一個(gè)從0開(kāi)始的遞增數(shù)字命名，Redis默認(rèn)支持16個(gè)數(shù)據(jù)庫(kù)（可以通過(guò)配置文件支持更多，無(wú)上限），可以通過(guò)配置databases來(lái)修改這一數(shù)字?？蛻舳伺cRedis建立連接后會(huì)自動(dòng)選擇0號(hào)數(shù)據(jù)庫(kù)，不過(guò)可以隨時(shí)使用SELECT命令更換數(shù)據(jù)庫(kù)

重要命令

命令執(zhí)行后輸出 (integer) 1，否則將輸出 (integer) 0

set
del
get
keys * 
lrange queue 0 -1  從第一個(gè)元素 (0) 到最后一個(gè)元素 (-1)BLPOP key1 [key2 ] timeout
移出并獲取列表的第一個(gè)元素， 如果列表沒(méi)有元素會(huì)阻塞列表直到等待超時(shí)或發(fā)現(xiàn)可彈出元素為止。

發(fā)布訂閱 pub/sub 模式

需要開(kāi)啟兩個(gè)redis-cli 客戶端

第一個(gè)：

subscribe chat 
訂閱該頻道

第二個(gè)：

publish chat "redis-test"
往chat頻道發(fā)消息

消息隊(duì)列

使用redis 進(jìn)行消息隊(duì)列，關(guān)鍵就是有消費(fèi)者、生產(chǎn)者的動(dòng)作，這里以python一段代碼進(jìn)行啟發(fā)：

#redis_aclient.py
from contextlib import asynccontextmanagerfrom django.conf import settings
from redis.asyncio import from_url as redis_from_urlclass RedisClient:def __init__(self):self._client = Nonedef connect(self):if self._client is None:# logger.info("Connecting to Redis...")self._client = redis_from_url(settings.REDIS_URL, decode_responses=True)@asynccontextmanagerasync def get_client(self):if self._client is None:self.connect()try:yield self._clientfinally:await self.close()async def close(self):if self._client:# logger.info("Shutting down Redis connection...")await self._client.aclose()# logger.info("Redis connection closed.")self._client = None# 單例模式 - 實(shí)例化 RedisClient
redis_client_manager = RedisClient()

調(diào)用生產(chǎn)者消費(fèi)者的邏輯

import asyncio
import json
import osimport django
from asgiref.sync import sync_to_asyncasync def generate_by_queue(answer_id, answer_data):"""向 Redis 隊(duì)列中推送任務(wù)"""task_data = {"answer_id": answer_id,"answer_data": answer_data}try:async with redis_client_manager.get_client() as redis_client:logger.info(f"Pushing task to queue answer_id: {answer_id}")await redis_client.rpush("ai_report_task_queue", json.dumps(task_data))except Exception as e:logger.info(f"Error pushing task to queue: {e}")async def process_tasks_by_ai_explain_answer():"""從 Redis 隊(duì)列中獲取任務(wù)并進(jìn)行處理"""try:async with redis_client_manager.get_client() as redis_client:while True:try:# 阻塞直到有任務(wù)出現(xiàn)task = await redis_client.blpop("queue")if task:task_data = json.loads(task[1])  # 解析任務(wù)數(shù)據(jù)answer_id = task_data.get("answer_id")logger.info(f"Processing task answer_id {answer_id}")answer_data = task_data.get("answer_data")ai_explain = await generate_ai_report(answer_data)await update_answer_record(answer_id, ai_explain)logger.info(f"Task answer_id {answer_id} processed")except Exception as e:logger.info(f"Error processing task {e}")except asyncio.CancelledError:logger.info("Task processing cancelled.")except Exception as e:logger.info(f"Unexpected error: {e}")if __name__ == "__main__":try:asyncio.run(process_tasks_by_ai_explain_answer())except KeyboardInterrupt:asyncio.run(redis_client_manager.close())