当前位置：首页 > article >正文

Redis 简单的消息队列

article 2024/9/30 17:21:40

使用redis 进行简单的队列很容易，不需要使用较为复杂的MQ队列，直接使用redis 进行，不过唯一不足的需要自己构造生产者消费者，这里使用while True的方法进行消费者操作

介绍

key-value 存储系统，是跨平台的非关系型数据库。Redis 通常被称为数据结构服务器，因为值（value）可以是字符串(String)、哈希(Hash)、列表(list)、集合(sets)和有序集合(sorted sets)等类型。

常用于缓存、消息队列、会话存储等应用场景。

**性能极高：**Redis 以其极高的性能而著称，能够支持每秒数十万次的读写操作24。这使得Redis成为处理高并发请求的理想选择，尤其是在需要快速响应的场景中，如缓存、会话管理、排行榜等。
**丰富的数据类型：**Redis 不仅支持基本的键值存储，还提供了丰富的数据类型，包括字符串、列表、集合、哈希表、有序集合等。这些数据类型为开发者提供了灵活的数据操作能力，使得Redis可以适应各种不同的应用场景。
**原子性操作：**Redis 的所有操作都是原子性的，这意味着操作要么完全执行，要么完全不执行。这种特性对于确保数据的一致性和完整性至关重要，尤其是在高并发环境下处理事务时。
**持久化：**Redis 支持数据的持久化，可以将内存中的数据保存到磁盘中，以便在系统重启后恢复数据。这为 Redis 提供了数据安全性，确保数据不会因为系统故障而丢失。
**支持发布/订阅模式：**Redis 内置了发布/订阅模式（Pub/Sub），允许客户端之间通过消息传递进行通信。这使得 Redis 可以作为消息队列和实时数据传输的平台。
**单线程模型：**尽管 Redis 是单线程的，但它通过高效的事件驱动模型来处理并发请求，确保了高性能和低延迟。单线程模型也简化了并发控制的复杂性。
**主从复制：**Redis 支持主从复制，可以通过从节点来备份数据或分担读请求，提高数据的可用性和系统的伸缩性。
**应用场景广泛：**Redis 被广泛应用于各种场景，包括但不限于缓存系统、会话存储、排行榜、实时分析、地理空间数据索引等。
**社区支持：**Redis 拥有一个活跃的开发者社区，提供了大量的文档、教程和第三方库，这为开发者提供了强大的支持和丰富的资源。
**跨平台兼容性：**Redis 可以在多种操作系统上运行，包括 Linux、macOS 和 Windows，这使得它能够在不同的技术栈中灵活部署。
• **支持 Lua 脚本：**Redis 支持使用 Lua 脚本来编写复杂的操作，这些脚本可以在服务器端执行，提供了更多的灵活性和强大的功能。

数据类型

Redis主要支持以下几种数据类型：

string（字符串）:

基本的数据存储单元，可以存储字符串、整数或者浮点数。
hash（哈希）：

一个键值对集合，可以存储多个字段。
list（列表）：

一个简单的列表，可以存储一系列的字符串元素。
set（集合）：

一个无序集合，可以存储不重复的字符串元素。
zset(sorted set：社群集合)：

相似集合，但是每个元素都有一个分数（score）关联。
位图（Bitmaps）：

基于操作字符串类型，可以对每个位进行。
超日志（HyperLogLogs）：

用于基本统计，可以提示集合中的唯一元素数量。
地理空间（Geospatial）：

用于存储断层信息。
发布/订阅（Pub/Sub）：

一种消息通信模式，允许客户端订阅消息通道，并接收发布到该通道的消息。
流（Streams）：

用于消息队列和日志存储，支持消息的持久化和时间排序。
模块（Modules）：

Redis支持动态加载模块，可以扩展Redis的功能。

String

string 是 redis 最基本的类型,string类型最大存储512MB

SET key value：设置键的值。
GET key：获取键的值。
INCR key：将键的值加 1。
DECR key：将键的值减 1。
APPEND key value：将值追加到键的值之后。

> set run "1"

Hash

Redis hash 是一个键值(key=>value)对集合，类似于一个小型的 NoSQL 数据库。

Redis hash 是一个 string 类型的 field 和 value 的映射表，hash 特别适合用于存储对象。

HSET key field value：设置哈希表中字段的值。
HGET key field：获取哈希表中字段的值。
HGETALL key：获取哈希表中所有字段和值。
HDEL key field：删除哈希表中的一个或多个字段。

最常用的还是string，其他用到了再说。

Redis支持多个数据库，并且每个数据库的数据是隔离的不能共享，并且基于单机才有，如果是集群就没有数据库的概念。

Redis是一个字典结构的存储服务器，而实际上一个Redis实例提供了多个用来存储数据的字典，客户端可以指定将数据存储在哪个字典中。这与我们熟知的在一个关系数据库实例中可以创建多个数据库类似，所以可以将其中的每个字典都理解成一个独立的数据库。

每个数据库对外都是一个从0开始的递增数字命名，Redis默认支持16个数据库（可以通过配置文件支持更多，无上限），可以通过配置databases来修改这一数字。客户端与Redis建立连接后会自动选择0号数据库，不过可以随时使用SELECT命令更换数据库

重要命令

命令执行后输出 (integer) 1，否则将输出 (integer) 0

set
del
get
keys * 
lrange queue 0 -1  从第一个元素 (0) 到最后一个元素 (-1)

BLPOP key1 [key2 ] timeout
移出并获取列表的第一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。

发布订阅 pub/sub 模式

在这里插入图片描述

需要开启两个redis-cli 客户端

第一个：

subscribe chat 
订阅该频道

第二个：

publish chat "redis-test"
往chat频道发消息

消息队列

使用redis 进行消息队列，关键就是有消费者、生产者的动作，这里以python一段代码进行启发：

#redis_aclient.py
from contextlib import asynccontextmanager

from django.conf import settings
from redis.asyncio import from_url as redis_from_url

class RedisClient:
    def __init__(self):
        self._client = None

    def connect(self):
        if self._client is None:
            # logger.info("Connecting to Redis...")
            self._client = redis_from_url(settings.REDIS_URL, decode_responses=True)

    @asynccontextmanager
    async def get_client(self):
        if self._client is None:
            self.connect()
        try:
            yield self._client
        finally:
            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

# 单例模式 - 实例化 RedisClient
redis_client_manager = RedisClient()

调用生产者消费者的逻辑

import asyncio
import json
import os

import django
from asgiref.sync import sync_to_async



async def generate_by_queue(answer_id, answer_data):
    """
    向 Redis 队列中推送任务
    """
    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 队列中获取任务并进行处理
    """
    try:
        async with redis_client_manager.get_client() as redis_client:
            while True:
                try:
                    # 阻塞直到有任务出现
                    task = await redis_client.blpop("queue")
                    if task:
                        task_data = json.loads(task[1])  # 解析任务数据
                        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())