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

网站空间和数据库的关系品牌网络推广

news 2025/7/13 0:15:36

网站空间和数据库的关系,品牌网络推广,北辰做网站的公司,wordpress合并目录一、Redis安装 1、下载安装包 2、解压文件 3、迁移文件夹 4、编译 5、管理redis文件 6、修改配置文件 7、启动Redis 8、将redis服务交给systemd管理二、Redis介绍 1、数据结构 ①字符串String ②列表List ③哈希Hash ④集合Set ⑤有序集合Sorted Set 2、…

一、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 redisclass 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 Nonedef 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 unittestclass 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()