linux安装redis及Python操作redis

目录

一、Redis安装

1、下载安装包

2、解压文件

3、迁移文件夹

4、编译

5、管理redis文件

6、修改配置文件

7、启动Redis

8、将redis服务交给systemd管理

二、Redis介绍

1、数据结构

①字符串String

②列表List

③哈希Hash

④集合Set

⑤有序集合Sorted Set

2、Redis特点

①高性能

②持久化

③分布式支持

三、Python连接Redis

四、总结

一、Redis安装

1、下载安装包

wget https://download.redis.io/releases/redis-6.2.6.tar.gz

2、解压文件

tar -zxvf redis-6.2.6.tar.gz

3、迁移文件夹

mv redis-6.2.6 /opt/redis

4、编译

cd /opt/redis
make
make install

5、管理redis文件

创建 bin 和 etc 文件夹

bin：用于存放可执行文件

etc：用于存放redis.conf

mkdir bin
mkdir etc

# 移动配置文件

mv redis.conf /opt/redis/etc/

# 移动可执行文件
cd src
mv mkreleasehdr.sh redis-benchmark redis-check-aof redis-check-rdb redis-cli redis-server /opt/redis/bin/
cd ../bin

6、修改配置文件

vim /opt/redis/etc/redis.conf


# 注释掉 bing 127.0.0.1，该项限制Redis只允许本地连接。将其注释掉后，Redis服务器将不再只监听本地回环地址，从而允许远程客户端连接到Redis服务器

# 取消requirepass foobared的注释，将foobared修改为密码
requirepass your_password

# 将appendonly 修改为 yes， appendonly配置项用于开启或关闭 AOF（Append Only File）持久化模式。将其设置为yes后，会开启 AOF 持久化。AOF 持久化模式会将每一个写命令追加到文件末尾，相比默认开启的 RDB 持久化，AOF 恢复的数据通常更完整，能在服务器故障恢复时尽可能保证数据的一致性和完整性
appendonly yes

# 将daemonize 修改为 yes，daemonize配置项用于控制Redis是否以守护进程的方式运行。将其设置为yes后，Redis会在后台运行，不会占用当前终端会话，方便在系统后台持续稳定地提供服务。
daemonize yes

# dir配置项用于指定 Redis 的数据存储目录。通过修改dir后面的路径，可将 Redis 的数据存储到指定的目录下，方便进行数据管理和备份等操作。
dir /your_data_dir

7、启动Redis

./redis-server /opt/redis/etc/redis.conf

进入客户端

./redis-cli

# 验证设置的密码
auth 密码

# 退出命令
exit

8、将redis服务交给systemd管理

编写 redis.service 

cd /etc/systemd/system
vim redis.service

写入以下内容：

[Unit]
Description=Redis
After=network.target

[Service]
Type=forking
PIDFile=/run/redis_6379.pid
ExecStart=/opt/redis/bin/redis-server /opt/redis/etc/redis.conf
ExecReload=/bin/kill -s HUP $MAINPID
Restart=always
ExecStop=/bin/kill -s QUIT $MAINPID
PrivateTmp=true

[Install]
WantedBy=multi-user.target

刷新并启动Redis

systemctl daemon-reload
systemctl start redis

二、Redis介绍

1、数据结构

①字符串String

最基本的数据结构，能存储任何形式的字符串，包括整数和浮点数（在 Redis 内部会进行转换）。例如，可以存储用户的姓名、文章的标题、计数器的值等。

②列表List

一个有序的字符串列表，可以在列表的两端进行插入（LPUSH和RPUSH）和弹出（LPOP和RPOP）操作，还可以获取指定范围的元素等。常用于消息队列、最新消息列表等场景。

③哈希Hash

一个键值对集合，其中的键和值都是字符串类型。适合存储对象的属性，如用户对象的姓名、年龄、性别等信息。

④集合Set

无序的、不包含重复元素的字符串集合。可以用于实现标签系统、好友关系等功能。

⑤有序集合Sorted Set

和集合类似，但每个元素都关联一个分数（Score），元素根据分数进行排序。常用于排行榜、优先级队列等场景。

2、Redis特点

①高性能

内存存储：数据存储在内存中，读写速度极快，能够在微秒级或者亚毫秒级完成读写操作，相比传统的基于磁盘的数据库，大大提高了数据访问效率。

单线程架构：采用单线程模型处理请求，避免了多线程的上下文切换开销，并且 Redis 的操作都是原子性的，这使得它在处理高并发场景下依然能够保持高性能和数据的一致性。

②持久化

RDB（Redis Database Backup）持久化：在指定的时间间隔内，将内存中的数据集快照写入磁盘。这种方式非常适合用于备份和灾难恢复，它生成的快照文件是紧凑的二进制文件，恢复速度快。

AOF（Append Only File）持久化：以日志的形式记录服务器所执行的所有写操作，在服务器启动时，通过重新执行这些写操作来恢复数据集。AOF 持久化方式提供了更好的数据安全性，可以设置不同的同步频率来平衡性能和数据安全性。

③分布式支持

主从复制（Master - Slave Replication）：支持一个主节点（Master）和多个从节点（Slave）的架构。主节点负责写操作，从节点负责复制主节点的数据，用于读操作，这样可以分担服务器的负载，提高系统的并发读写能力。

Redis Sentinel（哨兵）：用于监控主从节点的健康状态，当主节点出现故障时，能够自动将一个从节点升级为新的主节点，实现故障自动转移，提高系统的可用性。

Redis Cluster（集群）：通过分片（Sharding）的方式将数据分散存储在多个节点上，每个节点负责一部分数据的存储和处理，从而能够支持更大规模的数据存储和更高的并发访问量。

三、Python连接Redis

我将Python连接Redis的各类常用操作封装为了一个Class，使用的时候直接调用相关方法即可：

import redis

class RedisOperator:
    def __init__(self, host='localhost', port=6379, db=0, password='yj123'):
        """
        初始化Redis连接
        :param host: Redis服务器主机地址，默认为localhost
        :param port: Redis服务器端口号，默认为6379
        :param db: 要使用的数据库编号，默认为0
        :param password: 密码
        """
        self.r = redis.StrictRedis(host=host, port=port, db=db, password=password)

    # 字符串操作相关方法
    def set_string_value(self, key, value):
        """
        设置字符串类型的键值对
        :param key: 键
        :param value: 值
        :return: 设置操作是否成功
        """
        return self.r.set(key, value)

    def get_string_value(self, key):
        """
        获取字符串类型键对应的值
        :param key: 键
        :return: 对应的值，如果不存在则返回None
        """
        result = self.r.get(key)
        return result.decode('utf-8') if result else None

    # 列表操作相关方法
    def push_list_value(self, list_key, value, right=True):
        """
        向列表右侧/左侧添加元素
        :param list_key: 列表的键
        :param value: 要添加的元素
        :param right: 右侧
        :return: 列表长度
        """
        return self.r.rpush(list_key, value) if right else self.r.lpush(list_key, value)

    def get_list_range(self, list_key, start=0, end=-1):
        """
        获取列表指定范围内的元素
        :param list_key: 列表的键
        :param start: 起始索引
        :param end: 结束索引
        :return: 元素列表（字节类型元素组成的列表）
        """
        return self.r.lrange(list_key, start, end)

    # 哈希操作相关方法
    def set_hash_value(self, hash_key, field, value):
        """
        在哈希表中设置键值对
        :param hash_key: 哈希表的键
        :param field: 哈希表中的字段
        :param value: 对应的值
        :return: 成功设置的字段数量
        """
        return self.r.hset(hash_key, field, value)

    def get_hash_value(self, hash_key, field):
        """
        获取哈希表中指定字段的值
        :param hash_key: 哈希表的键
        :param field: 哈希表中的字段
        :return: 对应的值，如果不存在则返回None
        """
        result = self.r.hget(hash_key, field)
        return result.decode('utf-8') if result else None

    def get_multiple_hash_values(self, hash_key, fields):
        """
        获取哈希表中多个字段的值
        :param hash_key: 哈希表的键
        :param fields: 字段列表
        :return: 对应的值列表（字节类型元素组成的列表）
        """
        return self.r.hmget(hash_key, fields)

    # 集合操作相关方法
    def add_set_value(self, set_key, value):
        """
        向集合中添加元素
        :param set_key: 集合的键
        :param value: 要添加的元素
        :return: 如果元素是新添加的返回1，否则返回0
        """
        return self.r.sadd(set_key, value)

    def get_set_members(self, set_key):
        """
        获取集合中的所有元素
        :param set_key: 集合的键
        :return: 元素列表（字节类型元素组成的列表）
        """
        return list(self.r.smembers(set_key))

    # 有序集合操作相关方法
    def add_zset_value(self, zset_key, mapping):
        """
        向有序集合中添加元素（可批量添加，以字典形式传入元素和分数）
        :param zset_key: 有序集合的键
        :param mapping: 元素和对应分数的字典，例如 {'element1': score1, 'element2': score2}
        :return: 添加的元素数量
        """
        return self.r.zadd(zset_key, mapping)

    def get_zset_range(self, zset_key, start=0, end=-1, withscores=False):
        """
        获取有序集合指定范围内的元素
        :param zset_key: 有序集合的键
        :param start: 起始索引
        :param end: 结束索引
        :param withscores: 是否同时获取元素对应的分数，默认为False
        :return: 如果with_scores为False，返回元素列表（字节类型元素组成的列表）；如果with_scores为True，返回元素和分数的元组组成的列表
        """
        if withscores:
            return self.r.zrange(zset_key, start, end, withscores=withscores)
        return self.r.zrange(zset_key, start, end)

    def delete_key(self, key):
        """
        删除指定的键
        :param key: 要删除的键
        :return: 删除成功返回True，否则返回False
        """
        return self.r.delete(key)

以下使用unittest框架对RedisOperator类进行测试：

import unittest

class TestRedisOperator(unittest.TestCase):
    def setUp(self):
        self.redis_op = RedisOperator()

    # 测试字符串操作相关方法
    def test_set_string_value(self):
        result = self.redis_op.set_string_value("name", "James")
        self.assertEqual(isinstance(result, bool), True)

    def test_get_string_value(self):
        self.redis_op.set_string_value("name", "James")
        result = self.redis_op.get_string_value("name")
        self.assertEqual(result, "James")

    # 测试列表操作相关方法
    def test_push_list_value(self):
        result = self.redis_op.push_list_value("name_list", "James", right=True)
        self.assertEqual(isinstance(result, int), True)
        result = self.redis_op.push_list_value("name_list", "Bob", right=False)
        self.assertEqual(isinstance(result, int), True)

    def test_get_list_range(self):
        result = self.redis_op.get_list_range("name_list")
        self.assertEqual(isinstance(result, list), True)

    # 测试哈希操作相关方法
    def test_set_hash_value(self):
        result = self.redis_op.set_hash_value("hash_key1", "name", "James")
        self.assertEqual(isinstance(result, int), True)

    def test_get_hash_value(self):
        self.redis_op.set_hash_value("hash_key1", "name", "James")
        result = self.redis_op.get_hash_value("hash_key1", "name")
        self.assertEqual(result, "James")

    def test_get_multiple_hash_values(self):
        hash_key = "hash_key2"
        fields = ["field1", "field2"]
        values = ["value1", "value2"]
        for field, value in zip(fields, values):
            self.redis_op.set_hash_value(hash_key, field, value)
        result = self.redis_op.get_multiple_hash_values(hash_key, fields)
        self.assertEqual(isinstance(result, list), True)

    # 测试集合操作相关方法
    def test_add_set_value(self):
        result = self.redis_op.add_set_value("name_set", "James")
        self.assertEqual(isinstance(result, int), True)
        result = self.redis_op.add_set_value("name_set", "Bob")
        self.assertEqual(isinstance(result, int), True)

    def test_get_set_members(self):
        result = self.redis_op.get_set_members("name_set")
        self.assertEqual(isinstance(result, list), True)

    # 测试有序集合操作相关方法
    def test_add_zset_value(self):
        result = self.redis_op.add_zset_value("score_zset", {"James": 86, "Bob": 90, "Alice": 65})
        self.assertEqual(isinstance(result, int), True)

    def test_get_zset_range(self):
        result = self.redis_op.get_zset_range("score_zset")
        self.assertEqual(isinstance(result, list), True)

    def test_get_zset_range_with_scores(self):
        result = self.redis_op.get_zset_range("score_zset", withscores=True)
        self.assertEqual(isinstance(result, list), True)

    # 测试删除键操作
    def test_delete_key(self):
        for key in  ["name", "name_list", "hash_key1", "hash_key2", "name_set", "score_zset"]:
            result = self.redis_op.delete_key(key)
            self.assertEqual(isinstance(result, int), True)


if __name__ == '__main__':
    unittest.main()

四、总结

Redis是一个不错的缓存工具，可以作为应用程序和后端数据库之间的缓存层，把频繁访问的数据放在Redis，可以大大减少对后端数据库的访问压力。

Redis中的有序列表、Pub/Sub功能也可以用来实现简单的消息队列系统，例如用于异步处理任务、解耦系统组件等。

Redis中的有序集合可以实现排行榜功能，能够实时更新排名数据并快速查询排名情况。

Redis的原子特性可以用来实现计数器功能，如网站的访问量计数、用户点赞数计数等，能够保证在高并发情况下计数的准确性。