Redis 缓存数据库¶

# 方式一：APT安装（推荐）
sudo apt update
sudo apt install redis-server -y  # 安装Redis

# 启动服务
sudo systemctl start redis-server   # 启动
sudo systemctl enable redis-server  # 开机自启

# 验证
redis-cli ping  # 应返回 PONG

docker run -d \
    --name redis \
    -p 6379:6379 \
    redis:latest  # 启动Redis容器

# 连接
docker exec -it redis redis-cli

brew install redis  # Homebrew安装
brew services start redis  # 启动服务

pip install redis  # 安装Python Redis客户端

# 编辑配置文件
sudo vim /etc/redis/redis.conf

# 常改的配置项：
# bind 0.0.0.0          允许远程连接（默认只允许本地）
# requirepass your_pwd   设置密码
# maxmemory 256mb       最大内存限制
# maxmemory-policy allkeys-lru  内存满时删除策略（LRU最近最少使用）

# 重启生效
sudo systemctl restart redis-server

redis-cli  # 进入Redis命令行

# ===== String 字符串 =====
SET name "Alice"          # 设置键值
GET name                  # 获取值 → "Alice"
SET counter 100           # 设置数字
INCR counter              # 自增1 → 101
DECRBY counter 10         # 减10 → 91
SET session "abc" EX 3600 # 设置并指定过期时间3600秒
TTL session               # 查看剩余过期时间

# ===== Hash 哈希 =====
HSET user:1001 name "Bob" age 25 city "北京"  # 设置多个字段
HGET user:1001 name       # 获取单个字段 → "Bob"
HGETALL user:1001         # 获取所有字段
HDEL user:1001 city       # 删除字段

# ===== List 列表 =====
LPUSH queue "task1"       # 左端插入
RPUSH queue "task2"       # 右端插入
LPOP queue                # 左端弹出（消费）
LRANGE queue 0 -1         # 查看全部

# ===== Set 集合 =====
SADD tags "python" "redis" "linux"  # 添加标签
SMEMBERS tags             # 查看所有成员
SISMEMBER tags "python"   # 是否包含 → 1(是)
SINTER tags1 tags2        # 两个集合的交集

# ===== Sorted Set 有序集合 =====
ZADD leaderboard 100 "Alice" 95 "Bob" 88 "Charlie"  # 添加分数
ZREVRANGE leaderboard 0 2 WITHSCORES  # Top 3（降序）
ZRANK leaderboard "Bob"   # 排名（升序）
ZINCRBY leaderboard 5 "Bob"  # 给Bob加5分

# ===== 通用命令 =====
KEYS *                    # 查看所有键（生产环境慎用）
EXISTS name               # 键是否存在
DEL name                  # 删除键
EXPIRE name 60            # 设置60秒后过期
TYPE name                 # 查看数据类型

# redis_example.py
import redis  # 导入Redis客户端
import json

# 连接Redis
r = redis.Redis(
    host='localhost',  # Redis地址
    port=6379,         # 端口
    db=0,              # 数据库编号（0-15）
    decode_responses=True,  # 自动解码为字符串
)

# 测试连接
print(r.ping())  # True

# ===== 缓存模式 =====
def get_user(user_id: int) -> dict:
    """先查缓存，没有再查数据库"""
    cache_key = f"user:{user_id}"  # 缓存键

    # 尝试从Redis获取
    cached = r.get(cache_key)
    if cached:
        print("命中缓存")
        return json.loads(cached)  # 反序列化返回

    # 缓存没有，从数据库查
    print("查询数据库")
    user = {"id": user_id, "name": "Alice", "age": 25}  # 模拟数据库查询

    # 存入缓存，设置1小时过期
    r.setex(
        cache_key,
        3600,  # 过期时间（秒）
        json.dumps(user),  # 序列化为JSON字符串
    )
    return user

# ===== 计数器 =====
r.set("page_views", 0)  # 初始化
r.incr("page_views")    # 页面浏览+1
r.incr("page_views")    # 再+1
print(f"页面浏览: {r.get('page_views')}")  # 2

# ===== 排行榜 =====
r.zadd("scores", {"Alice": 100, "Bob": 95, "Charlie": 88})  # 添加分数
top3 = r.zrevrange("scores", 0, 2, withscores=True)  # Top 3
print(f"排行榜: {top3}")  # [('Alice', 100.0), ('Bob', 95.0), ...]

# ===== 分布式锁 =====
import time

lock_key = "lock:resource"
if r.set(lock_key, "locked", nx=True, ex=10):  # nx=不存在才设置, ex=10秒超时
    try:
        print("获得锁，执行任务...")
        time.sleep(2)  # 模拟耗时操作
    finally:
        r.delete(lock_key)  # 释放锁
else:
    print("未获得锁，资源被占用")

# Pipeline把多个命令打包一次发送，减少网络往返
pipe = r.pipeline()  # 创建Pipeline
for i in range(1000):
    pipe.set(f"key:{i}", f"value:{i}")  # 缓冲命令
results = pipe.execute()  # 一次性执行所有命令
# 比逐条执行快10倍以上

# publisher.py（发布者）
r.publish("news", "重大新闻：Python 4.0发布!")  # 发布消息到news频道

# subscriber.py（订阅者）
pubsub = r.pubsub()  # 创建订阅对象
pubsub.subscribe("news")  # 订阅news频道
for message in pubsub.listen():  # 监听消息
    if message["type"] == "message":
        print(f"收到: {message['data']}")

# 生产消息
r.xadd("events", {  # 往events流添加消息
    "user_id": "1001",
    "action": "login",
    "ip": "192.168.1.1",
})

# 消费消息（消费者组）
r.xgroup_create("events", "my-group", id="0", mkstream=True)  # 创建消费者组
messages = r.xreadgroup(
    "my-group", "consumer-1",  # 组名和消费者名
    {"events": ">"},  # ">"表示读取新消息
    count=10,  # 最多读10条
)
for stream, msgs in messages:
    for msg_id, data in msgs:
        print(f"消息 {msg_id}: {data}")
        r.xack("events", "my-group", msg_id)  # 确认消费

# Redis 8.0内置向量搜索，可用于RAG/语义搜索
import numpy as np

# 创建向量索引
r.execute_command(
    'FT.CREATE', 'doc_idx',  # 索引名
    'ON', 'HASH',  # 索引Hash类型的数据
    'PREFIX', '1', 'doc:',  # 索引doc:开头的键
    'SCHEMA',
    'content', 'TEXT',  # 文本字段
    'embedding', 'VECTOR', 'FLAT', '6',  # 向量字段
    'TYPE', 'FLOAT32', 'DIM', '384', 'DISTANCE_METRIC', 'COSINE',
)

# 存入文档和向量
embedding = np.random.rand(384).astype(np.float32).tobytes()
r.hset("doc:1", mapping={
    "content": "Redis是高性能内存数据库",
    "embedding": embedding,
})

# ===== 连接 =====
redis-cli                      # 本地连接
redis-cli -h host -p 6379 -a pwd  # 远程连接

# ===== 数据类型命令速查 =====
# String: SET GET INCR DECR SETEX SETNX MSET MGET
# Hash:   HSET HGET HGETALL HDEL HEXISTS HINCRBY
# List:   LPUSH RPUSH LPOP RPOP LRANGE LLEN
# Set:    SADD SREM SMEMBERS SINTER SUNION SDIFF
# ZSet:   ZADD ZREM ZRANGE ZREVRANGE ZRANK ZSCORE

# ===== 管理命令 =====
INFO                   # 服务器信息
DBSIZE                 # 当前数据库键数
FLUSHDB                # 清空当前数据库
FLUSHALL               # 清空所有数据库（危险！）
SAVE                   # 手动RDB快照
BGSAVE                 # 后台RDB快照
MONITOR                # 实时监控所有命令

# ===== 内存淘汰策略 =====
# noeviction     不淘汰，满了就报错
# allkeys-lru    所有键LRU淘汰（最常用）
# volatile-lru   只淘汰有过期时间的键
# allkeys-random 随机淘汰
# volatile-ttl   淘汰TTL最小的键