品牌网站建设找哪家,崂山区城市规划建设局网站,三明企业网站建设,伊犁建设网站公司使用redis 进行简单的队列很容易#xff0c;不需要使用较为复杂的MQ队列#xff0c;直接使用redis 进行#xff0c;不过唯一不足的需要自己构造生产者消费者#xff0c;这里使用while True的方法进行消费者操作 目录 介绍数据类型StringHash 重要命令消息队列 介绍
key-v…使用redis 进行简单的队列很容易不需要使用较为复杂的MQ队列直接使用redis 进行不过唯一不足的需要自己构造生产者消费者这里使用while True的方法进行消费者操作 目录 介绍数据类型StringHash 重要命令消息队列 介绍
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 1Hash
Redis hash 是一个键值(keyvalue)对集合类似于一个小型的 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 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_responsesTrue)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# 单例模式 - 实例化 RedisClient
redis_client_manager RedisClient()
调用生产者消费者的逻辑
import asyncio
import json
import osimport django
from asgiref.sync import sync_to_asyncasync 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(fPushing 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(fError 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(fProcessing 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(fTask answer_id {answer_id} processed)except Exception as e:logger.info(fError processing task {e})except asyncio.CancelledError:logger.info(Task processing cancelled.)except Exception as e:logger.info(fUnexpected error: {e})if __name__ __main__:try:asyncio.run(process_tasks_by_ai_explain_answer())except KeyboardInterrupt:asyncio.run(redis_client_manager.close())
