Redis 是一个开源的、高性能的键值对（key-value）内存数据库，也常被称为数据结构服务器。它支持多种数据结构，具有速度快、功能丰富、可扩展性强等特点，广泛应用于缓存、会话存储、消息队列、实时分析等场景。

Redis 的核心特点

1. 内存存储：数据主要存储在内存中，读写速度极快（毫秒级响应）。
2. 多数据结构：支持字符串（String）、哈希（Hash）、列表（List）、集合（Set）、有序集合（Sorted Set）等多种数据结构。
3. 持久化：可以通过 RDB（快照）和 AOF（日志）两种方式将数据持久化到磁盘，避免数据丢失。
4. 高并发：单线程模型避免了线程切换开销，同时支持异步 I/O，能高效处理大量并发请求。
5. 支持集群：通过 Redis Cluster 实现分布式部署，提供高可用和水平扩展能力。
6. 丰富的功能：支持事务、发布订阅、Lua 脚本、过期键自动删除等功能。

常用数据结构及操作示例

1. 字符串（String）
   最基本的键值对类型，可存储文本或二进制数据。

   SET name "Redis"    # 设置键值
   GET name            # 获取值 → "Redis"
   INCR counter        # 自增（适用于计数器）
   

2. 哈希（Hash）
   适合存储对象，如用户信息、商品属性等。

   HSET user:1 name "Alice" age 30  # 设置哈希字段
   HGET user:1 name                 # 获取字段值 → "Alice"
   HGETALL user:1                   # 获取所有字段和值
   

3. 列表（List）
   有序的字符串集合，可用于消息队列、最新消息展示等。

   LPUSH fruits "apple"    # 左侧插入元素
   RPUSH fruits "banana"   # 右侧插入元素
   LRANGE fruits 0 -1      # 获取所有元素 → ["apple", "banana"]
   

4. 集合（Set）
   无序且唯一的元素集合，支持交集、并集等操作。

   SADD tags "python" "redis"  # 添加元素
   SMEMBERS tags               # 获取所有元素
   SINTER set1 set2            # 求两个集合的交集
   

5. 有序集合（Sorted Set）
   类似集合，但每个元素关联一个分数，可按分数排序，适合排行榜等场景。

   ZADD scores 90 "Alice" 85 "Bob"  # 添加元素及分数
   ZRANGE scores 0 -1 WITHSCORES    # 按分数升序排列
   ZREVRANGE scores 0 2             # 取前三名（降序）
   

Python 操作 Redis 示例

使用 redis-py 库可以方便地在 Python 中操作 Redis：

import redis

# 连接 Redis 服务器
r = redis.Redis(host='localhost', port=6379, db=0)

# 字符串操作
r.set('name', 'Redis')
print(r.get('name').decode('utf-8'))  # 输出：Redis

# 哈希操作
r.hset('user:1', mapping={'name': 'Alice', 'age': '30'})
print(r.hget('user:1', 'name').decode('utf-8'))  # 输出：Alice

# 列表操作
r.lpush('fruits', 'apple', 'banana')
print(r.lrange('fruits', 0, -1))  # 输出：[b'banana', b'apple']

应用场景

• 缓存：减轻数据库压力，存储热点数据（如商品详情、用户信息）。
• 会话存储：存储用户登录状态（如 Session ID）。
• 消息队列：基于 List 实现简单的生产者-消费者模型。
• 计数器：如文章阅读量、点赞数（利用 INCR 命令）。
• 排行榜：基于 Sorted Set 实现实时排名（如游戏分数榜）。

Redis 凭借其高性能和灵活性，成为现代应用架构中不可或缺的组件之一。

使用Redis实现分布式锁

使用Redis实现分布式锁是分布式系统中解决并发问题的常见方案，其核心思想是通过Redis的原子操作确保同一时刻只有一个客户端能获取到锁。

分布式锁的核心要求

1. 互斥性：同一时刻只能有一个客户端持有锁。
2. 安全性：避免死锁（如客户端崩溃后锁无法释放）。
3. 可用性：Redis节点故障时仍能正常工作（可通过集群实现）。
4. 原子性：锁的获取和释放操作必须是原子的。

基于Redis的分布式锁实现

1. 获取锁（关键命令）
   使用 SET key value NX PX timeout 命令：

   • NX：仅当键不存在时才设置（确保互斥）。
   • PX timeout：设置过期时间（避免死锁）。
   • value：使用唯一标识（如UUID），确保释放锁时不会误删其他客户端的锁。

2. 释放锁（关键命令）
   通过Lua脚本实现原子操作，先判断锁的持有者是否为当前客户端，再删除锁：

   if redis.call('get', KEYS[1]) == ARGV[1] then
     return redis.call('del', KEYS[1])
   else
     return 0
   end
   

Python实现示例

使用 redis-py 库实现分布式锁：

import redis
import uuid
import time

class RedisDistributedLock:
    def __init__(self, redis_client, lock_key, expire_time=10):
        """
        初始化分布式锁
        :param redis_client: Redis客户端连接
        :param lock_key: 锁的键名
        :param expire_time: 锁的过期时间（秒），防止死锁
        """
        self.redis = redis_client
        self.lock_key = lock_key
        self.expire_time = expire_time * 1000  # 转换为毫秒
        self.lock_value = str(uuid.uuid4())  # 唯一标识，用于释放锁
        self.is_locked = False  # 是否持有锁

    def acquire(self, retry_times=3, retry_interval=0.5):
        """
        获取锁
        :param retry_times: 获取失败时重试次数
        :param retry_interval: 重试间隔（秒）
        :return: 是否获取成功
        """
        for _ in range(retry_times):
            # 使用SET命令的NX和PX参数实现原子性获取锁
            result = self.redis.set(
                self.lock_key,
                self.lock_value,
                nx=True,  # 仅当键不存在时才设置
                px=self.expire_time  # 设置过期时间
            )
            if result:
                self.is_locked = True
                return True
            # 重试前等待
            time.sleep(retry_interval)
        return False

    def release(self):
        """释放锁，通过Lua脚本确保原子性"""
        if not self.is_locked:
            return False
        
        # Lua脚本：判断锁是否为当前客户端持有，是则删除
        lua_script = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        """
        # 执行脚本
        result = self.redis.eval(
            lua_script,
            1,  # 键的数量
            self.lock_key,  # KEYS[1]
            self.lock_value  # ARGV[1]
        )
        self.is_locked = False
        return result == 1

    def __del__(self):
        """析构函数，确保锁被释放"""
        if self.is_locked:
            self.release()

# 使用示例
if __name__ == "__main__":
    # 连接Redis
    redis_client = redis.Redis(host='localhost', port=6379, db=0)
    
    # 创建锁实例
    lock = RedisDistributedLock(redis_client, "distributed_lock", expire_time=10)
    
    try:
        # 获取锁
        if lock.acquire():
            print("获取锁成功，执行临界区操作...")
            # 模拟业务逻辑
            time.sleep(5)
        else:
            print("获取锁失败")
    finally:
        # 释放锁
        if lock.release():
            print("释放锁成功")
        else:
            print("释放锁失败（可能已过期）")
    

关键注意事项

1. 过期时间设置：需根据业务执行时间合理设置，避免锁提前过期导致并发问题。
2. 锁续期：对于长耗时任务，可通过定时线程延长锁的过期时间（如Redisson的"看门狗"机制）。
3. Redis集群：单节点Redis可能存在单点故障，生产环境建议使用Redis Cluster，并通过Redlock算法提高可靠性。
4. 避免误删锁：必须通过唯一标识（如UUID）验证锁的持有者，防止释放其他客户端的锁。

成熟方案推荐

实际开发中，推荐使用成熟的分布式锁库（如Redisson），其已封装了锁续期、集群支持等高级特性，减少手动实现的风险。