企业官网建设流程全解析

目录

引言

Redis作为一种高性能的键值存储系统,它的作用非常强大。它在缓存这个场景里很重要。它在会话存储这种场合下也很关键。它还经常用来搭建消息队列。现在有很多人想要用到Redis那些很厉害的的功能。为了方便大家把功能用起来更简单。这篇文章打算讲得很详细。它会告诉大家怎么进行操作Redis。

一、关于与Redis的第一次相遇, 我们在1.1章节来解释为什么会选择使用Redis。

Redis它是一款开源的内存数据结构存储产品, 用户可以用它来当数据库用也可以用它来做缓存, 还可以拿来当作消息代理来处理事情, 这个软件支持好多种数据结构的玩法, 像什么字符串啊哈希啊列表啊集合啊这些结构都支持得了, 同时它还提供了把数据存储下来的持久化功能。

1.步骤2, 去安装叫做redis-py的那个东西。

redis-py这个库是与Redis进行交互操作的官方所推荐的一款库, 安装这个库是非常简单的。

pip install redis

二、进行Redis的连接操作, 需要掌握基础的知识以及高级的配置方法, 其中第二小点讲述的是关于基础的连接部分。

建立与Redis的基本关联。

import redis

​

# 最简单连接（默认localhost:6379，数据库0）

r = redis.Redis()

print("成功连接到Redis！" if r.ping() else "连接失败")

2.第二种情况就是需要带有一些参数进行连接的。

在大家实际搞生产和干活的环境里面, 通常情况下是需要用到比刚才说的更多的种种参数的, 所以说:

import redis

​

try:

r = redis.Redis(

host='your_redis_host', # Redis服务器地址

port=6379, # 端口，默认6379

password='your_password', # 认证密码

db=0, # 数据库索引（0-15）

decode_responses=True, # 自动将字节数据解码为字符串

socket_connect_timeout=5, # 连接超时时间（秒）

socket_timeout=5 # 操作超时时间（秒）

)

# 测试连接

r.ping()

print("成功连接到Redis！

")

except redis.exceptions.ConnectionError as e:

print(f"连接失败: {e}")

关键参数解析：

2.通过使用连接池来提高性能。

在高并发的情况下面对它, 选择去使用连接池这一种方式就能够让性能得到非常明显的提升效果。

import redis

​

# 创建连接池

pool = redis.ConnectionPool(

host='localhost',

port=6379,

password='your_password',

db=0,

decode_responses=True,

max_connections=10 # 连接池最大连接数

)

​

# 从连接池获取连接

r = redis.Redis(connection_pool=pool)

​

# 使用连接

r.set('pool_test', '连接池工作正常')

value = r.get('pool_test')

print(value) # 输出：连接池工作正常

连接池的好处就是让连接能够重复使用, 这样就不需要去频繁地创建新的连接和销毁旧有的连接, 从而省去了很多不必要的开销, 这个特点在高并发的情况下表现得尤为突出, 所以它特别适合像Web应用这种对并发量要求较高的场景。

三、Redis数据类型操作详解3.1 字符串（）

字符串类型是Redis里面最基础的那种数据结构。

# 基本设置和获取

r.set('username', 'john_doe')

r.setex('temp_session', 3600, 'session_data') # 带过期时间（秒）

​

# 批量操作

r.mset({'key1': 'value1', 'key2': 'value2', 'key3': 'value3'})

values = r.mget('key1', 'key2', 'key3')

​

# 数字操作

r.set('counter', 10)

r.incr('counter') # 增加1

r.incrby('counter', 5) # 增加5

r.decr('counter') # 减少1

​

print(r.get('counter')) # 输出：15

3.对于数字2而言, 我们需要执行哈希这个操作。

哈希这个方法, 它挺适合拿来存储那些对象的。

# 设置哈希值

r.hset('user:100', 'name', 'Alice')

r.hset('user:100', 'age', 30)

# 或使用mapping参数批量设置

r.hset('user:101', mapping={'name': 'Bob', 'age': 25, 'city': 'New York'})

​

# 获取数据

name = r.hget('user:100', 'name')

all_data = r.hgetall('user:101') # 获取所有字段

​

print(f"用户100姓名: {name}")

print(f"用户101全部数据: {all_data}")

3.也就是第3部分, 这部分内容是关于列表的具体讲解。

Redis当中的列表这个数据结构, 它是依靠链表来进行实现的, 所以它是非常适合用来做队列或者是栈这样的结构的:

# 列表操作

r.lpush('tasks', 'task1', 'task2', 'task3') # 左侧插入

r.rpush('tasks', 'task4') # 右侧插入

​

# 获取元素

first_task = r.lpop('tasks') # 左侧弹出

last_task = r.rpop('tasks') # 右侧弹出

​

# 获取范围元素

all_tasks = r.lrange('tasks', 0, -1) # 获取所有元素

​

print(f"第一个任务: {first_task}")

print(f"所有任务: {all_tasks}")

3.第四点是关于集合。

# 集合 - 无序且唯一

r.sadd('tags', 'python', 'redis', 'database', 'python') # 重复元素自动去重

​

tags = r.smembers('tags')

print(f"标签集合: {tags}") # 输出: {'python', 'redis', 'database'}

​

3.第五种是有序集合这个内容, 它在分类里面会被叫做是Sets。

# 有序集合 - 带分数排序

r.zadd('leaderboard', {'Alice': 100, 'Bob': 90, 'Charlie': 120})

top_players = r.zrevrange('leaderboard', 0, 1, withscores=True) # 降序排列

​

print(f"排行榜前两名: {top_players}")

四、关于高级功能与实战技巧的这一部分内容, 也就是第四点一章节, 它所探讨的核心主题涉及到数据过期以及自动清理的相关事宜。

Redis这个工具可以设置一个键的生存时间, 这样做就可以实现数据的自动过期。

# 设置键的过期时间

r.setex('session_token', 3600, 'encrypted_token_data') # 1小时后过期

r.expire('existing_key', 600) # 设置已有键的过期时间（10分钟）

# 检查剩余时间

ttl = r.ttl('session_token') # 获取剩余生存时间（秒）

print(f"会话令牌剩余生存时间: {ttl}秒")

# 监控数据生命周期

import time

r.set('temp_data', '重要但会消失的数据', ex=60) # 60秒后自动删除

while True:

remaining = r.ttl('temp_data')

if remaining > 0:

print(f"数据剩余寿命：{remaining}秒")

time.sleep(10)

else:

print("数据已自然消亡")

break

4.2 管道（）批量操作

大家在使用管道这个办法的时候, 能够非常明确地看到, 它会对批量操作的性能提升起到一个非常显著的作用。

import redis

r = redis.Redis(decode_responses=True)

# 创建管道

pipe = r.pipeline()

# 添加多个命令到管道

for i in range(1000):

pipe.set(f'key:{i}', f'value:{i}')

# 一次性执行所有命令

results = pipe.execute()

print(f"批量插入了{len(results)}条数据")

管道技术能够通过减少在网络往返上面所花费的次数, 从而提升系统整体的性能表现, 这种特性使得它在面对需要批量插入大量数据的应用场景时, 显得格外适合。

4.第3点, 事务这个东西的作用在于能够对原子性进行保证。

关于Redis事务这一事物, 它是能够去确保让那一连串的命令得到一个原子性的被执行效果的。

# 基本事务

pipe = r.pipeline(transaction=True) # 开启事务模式

try:

pipe.multi() # 开始事务

pipe.set('account:A', 100)

pipe.set('account:B', 100)

pipe.execute() # 执行事务

print("事务执行成功")

except redis.exceptions.RedisError as e:

print(f"事务执行失败: {e}")

# 使用WATCH实现乐观锁

def transfer_funds(source, destination, amount):

with r.pipeline() as pipe:

try:

# 监视源账户键

pipe.watch(source)

# 检查余额是否足够

current_balance = int(r.get(source) or 0)

if current_balance < amount:

pipe.unwatch()

print("余额不足")

return False

# 开始事务

pipe.multi()

pipe.decrby(source, amount)

pipe.incrby(destination, amount)

pipe.execute()

print("转账成功")

return True

except redis.WatchError:

print("账户余额已被其他操作修改，转账失败")

return False

# 测试转账

r.set('account:A', 500)

r.set('account:B', 500)

transfer_funds('account:A', 'account:B', 100)

4.4, 发布模式或者是订阅模式。

Redis这个软件, 它是支持那种发布和订阅消息的模式的:

# 订阅者

import threading

def subscriber(channel_name):

pubsub = r.pubsub()

pubsub.subscribe(channel_name)

for message in pubsub.listen():

if message['type'] == 'message':

print(f"收到消息: {message['data']}")

# 启动订阅线程

thread = threading.Thread(target=subscriber, args=('news',))

thread.daemon = True

thread.start()

# 发布者

r.publish('news', 'Hello, subscribers!

')

r.publish('news', '这是第二条消息')

五、实战应用场景5.1 缓存实现

def get_user_profile(user_id, cache_timeout=3600):

# 尝试从缓存获取

cache_key = f"user_profile:{user_id}"

cached_data = r.get(cache_key)

if cached_data:

print("从缓存获取数据")

return cached_data

# 缓存不存在，从数据库获取

print("从数据库获取数据")

user_data = f"用户{user_id的详细信息}" # 模拟数据库查询

# 存入缓存

r.setex(cache_key, cache_timeout, user_data)

return user_data

5.2 会话存储

def create_user_session(user_id):

session_id = f"session:{user_id}"

session_data = {

'user_id': user_id,

'username': 'john_doe',

'last_login': '2023-10-01 12:00:00'

}

# 会话保存2小时

r.hset(session_id, mapping=session_data)

r.expire(session_id, 7200)

return session_id

​

def get_session(session_id):

return r.hgetall(session_id)

5.第3部分, 关于限流器的相关内容。

def rate_limiter(user_id, action, limit=10, window=60):

key = f"rate_limit:{user_id}:{action}"

current = r.get(key)

if current and int(current) >= limit:

return False # 超过限制

pipe = r.pipeline()

pipe.incr(key)

pipe.expire(key, window) # 设置过期时间

pipe.execute()

return True # 在限制内

六、性能优化