Python Redis哈希案例如何读写哈希数据

wen python案例 2026-06-30 197

本文目录导读：

基础操作案例
完整业务案例：用户管理系统
高级操作案例
实际应用场景：购物车实现
性能优化和注意事项

我来给你介绍 Python 操作 Redis 哈希数据的完整案例。

基础操作案例

安装依赖

pip install redis

基本读写示例

import redis
# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# 1. 设置哈希数据
# 方法1：单个设置
r.hset("user:1001", "name", "张三")
r.hset("user:1001", "age", 25)
r.hset("user:1001", "email", "zhangsan@example.com")
# 方法2：批量设置
r.hset("user:1002", mapping={
    "name": "李四",
    "age": 30,
    "email": "lisi@example.com",
    "city": "北京"
})
# 2. 读取哈希数据
# 获取单个字段
name = r.hget("user:1001", "name")
print(f"姓名: {name}")  # 张三
# 获取多个字段
fields = r.hmget("user:1001", ["name", "age", "email"])
print(f"多个字段: {fields}")  # ['张三', '25', 'zhangsan@example.com']
# 获取所有字段和值
all_data = r.hgetall("user:1001")
print(f"所有数据: {all_data}")  # {'name': '张三', 'age': '25', 'email': 'zhangsan@example.com'}
# 获取所有字段名
keys = r.hkeys("user:1001")
print(f"所有字段: {keys}")  # ['name', 'age', 'email']
# 获取所有值
values = r.hvals("user:1001")
print(f"所有值: {values}")  # ['张三', '25', 'zhangsan@example.com']

完整业务案例：用户管理系统

import redis
import json
from datetime import datetime
import time
class UserManager:
    def __init__(self):
        self.r = redis.Redis(
            host='localhost',
            port=6379,
            db=0,
            decode_responses=True
        )
        self.user_prefix = "user:"
    def create_user(self, user_id, user_info):
        """创建用户"""
        key = f"{self.user_prefix}{user_id}"
        # 检查用户是否已存在
        if self.r.exists(key):
            print(f"用户 {user_id} 已存在")
            return False
        # 添加创建时间
        user_info['created_at'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        user_info['updated_at'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        # 批量设置哈希字段
        self.r.hset(key, mapping=user_info)
        print(f"用户 {user_id} 创建成功")
        return True
    def get_user(self, user_id, fields=None):
        """获取用户信息"""
        key = f"{self.user_prefix}{user_id}"
        if fields:
            # 获取指定字段
            return self.r.hmget(key, fields)
        else:
            # 获取所有信息
            return self.r.hgetall(key)
    def update_user(self, user_id, update_info):
        """更新用户信息"""
        key = f"{self.user_prefix}{user_id}"
        # 检查用户是否存在
        if not self.r.exists(key):
            print(f"用户 {user_id} 不存在")
            return False
        # 添加更新时间
        update_info['updated_at'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        # 更新字段
        self.r.hset(key, mapping=update_info)
        print(f"用户 {user_id} 更新成功")
        return True
    def delete_user(self, user_id):
        """删除用户"""
        key = f"{self.user_prefix}{user_id}"
        if self.r.delete(key):
            print(f"用户 {user_id} 删除成功")
            return True
        print(f"用户 {user_id} 不存在")
        return False
    def get_user_count(self):
        """获取用户总数"""
        pattern = f"{self.user_prefix}*"
        keys = self.r.keys(pattern)
        data_info = self.r.info()
        return len(keys)
    def list_all_users(self):
        """列出所有用户"""
        pattern = f"{self.user_prefix}*"
        keys = self.r.keys(pattern)
        users = []
        for key in keys:
            user_data = self.r.hgetall(key)
            user_data['user_id'] = key.replace(self.user_prefix, '')
            users.append(user_data)
        return users
# 使用示例
def user_management_demo():
    um = UserManager()
    # 1. 创建用户
    user1 = {
        'name': '张三',
        'age': 28,
        'email': 'zhangsan@example.com',
        'phone': '13800138001',
        'address': '北京市海淀区',
        'role': 'admin'
    }
    user2 = {
        'name': '李四',
        'age': 25,
        'email': 'lisi@example.com', 
        'phone': '13800138002',
        'address': '上海市浦东新区',
        'role': 'user'
    }
    um.create_user(1001, user1)
    um.create_user(1002, user2)
    # 2. 读取用户信息
    print("\n=== 读取用户信息 ===")
    # 获取所有信息
    user_info = um.get_user(1001)
    print(f"用户1001完整信息: {user_info}")
    # 获取指定字段
    specific_fields = um.get_user(1001, ['name', 'email', 'phone'])
    print(f"指定字段: {specific_fields}")
    # 3. 更新用户信息
    print("\n=== 更新用户信息 ===")
    update_info = {
        'age': 29,
        'phone': '13900139001',
        'hobby': '编程'
    }
    um.update_user(1001, update_info)
    updated_info = um.get_user(1001)
    print(f"更新后的信息: {updated_info}")
    # 4. 用户计数
    print(f"\n用户总数: {um.get_user_count()}")
    # 5. 列出所有用户
    print("\n=== 所有用户列表 ===")
    all_users = um.list_all_users()
    for user in all_users:
        print(f"用户{user['user_id']}: {user}")
# 运行示例
user_management_demo()

高级操作案例

import redis
import time
import random
class HashOperations:
    def __init__(self):
        self.r = redis.Redis(
            host='localhost',
            port=6379,
            db=0,
            decode_responses=True
        )
    def counter_example(self):
        """计数器示例 - 使用哈希存储多个计数器"""
        key = "page:views"
        # 初始化计数器
        self.r.hset(key, "homepage", 0)
        self.r.hset(key, "about", 0)
        self.r.hset(key, "contact", 0)
        # 模拟页面访问
        for _ in range(10):
            page = random.choice(["homepage", "about", "contact"])
            self.r.hincrby(key, page, 1)
        # 获取所有页面访问量
        views = self.r.hgetall(key)
        print(f"页面访问统计: {views}")
        # 获取单个页面访问量
        home_views = self.r.hget(key, "homepage")
        print(f"首页访问量: {home_views}")
    def self_definiendo_fields(self):
        """动态字段操作"""
        key = "game:player:1001"
        # 动态添加字段
        stats = {
            'level': 10,
            'exp': 5000,
            'health': 100,
            'mana': 50,
            'weapon': '剑',
            'armor': '盾牌'
        }
        self.r.hset(key, mapping=stats)
        # 检查字段是否存在
        if self.r.hexists(key, 'weapon'):
            print(f"玩家拥有武器: {self.r.hget(key, 'weapon')}")
        # 获取字段数量
        fields_count = self.r.hlen(key)
        print(f"玩家属性数量: {fields_count}")
        # 删除字段
        self.r.hdel(key, 'armor')
        print(f"删除后属性: {self.r.hgetall(key)}")
    def batch_operations(self):
        """批量操作示例"""
        pipeline = self.r.pipeline()
        # 批量设置多个哈希
        for i in range(5):
            user_key = f"student:{i}"
            pipeline.hset(user_key, mapping={
                'name': f'学生{i}',
                'score': random.randint(60, 100),
                'class': f'班级{i%3+1}'
            })
        pipeline.execute()
        # 批量获取
        keys = [f"student:{i}" for i in range(5)]
        pipeline = self.r.pipeline()
        for key in keys:
            pipeline.hgetall(key)
        results = pipeline.execute()
        for i, result in enumerate(results):
            print(f"学生{i}: {result}")
# 运行高级操作示例
def advanced_demo():
    ho = HashOperations()
    print("=== 计数器示例 ===")
    ho.counter_example()
    print("\n=== 动态字段操作 ===")
    ho.self_definiendo_fields()
    print("\n=== 批量操作 ===")
    ho.batch_operations()
advanced_demo()

实际应用场景：购物车实现

import redis
import json
class ShoppingCart:
    def __init__(self):
        self.r = redis.Redis(
            host='localhost',
            port=6379,
            db=1,  # 使用不同的数据库
            decode_responses=True
        )
        self.cart_prefix = "cart:"
    def add_item(self, user_id, product_id, quantity=1, product_info=None):
        """添加商品到购物车"""
        key = f"{self.cart_prefix}{user_id}"
        # 检查商品是否已在购物车
        if self.r.hexists(key, product_id):
            # 增加数量
            current_qty = int(self.r.hget(key, product_id))
            new_qty = current_qty + quantity
            self.r.hset(key, product_id, new_qty)
        else:
            # 新商品，设置数量
            self.r.hset(key, product_id, quantity)
        print(f"已添加 {product_id} 到购物车，数量: {quantity}")
    def remove_item(self, user_id, product_id):
        """从购物车移除商品"""
        key = f"{self.cart_prefix}{user_id}"
        if self.r.hexists(key, product_id):
            self.r.hdel(key, product_id)
            print(f"已移除 {product_id} 从购物车")
        else:
            print(f"购物车中没有 {product_id}")
    def update_quantity(self, user_id, product_id, quantity):
        """更新商品数量"""
        key = f"{self.cart_prefix}{user_id}"
        if quantity <= 0:
            self.remove_item(user_id, product_id)
        else:
            self.r.hset(key, product_id, quantity)
            print(f"已更新 {product_id} 数量为 {quantity}")
    def get_cart(self, user_id):
        """获取购物车内容"""
        key = f"{self.cart_prefix}{user_id}"
        cart_items = self.r.hgetall(key)
        if cart_items:
            print(f"\n用户 {user_id} 的购物车:")
            for product_id, quantity in cart_items.items():
                print(f"  商品: {product_id}, 数量: {quantity}")
        else:
            print(f"购物车为空")
        return cart_items
    def clear_cart(self, user_id):
        """清空购物车"""
        key = f"{self.cart_prefix}{user_id}"
        self.r.delete(key)
        print(f"已清空用户 {user_id} 的购物车")
    def get_cart_count(self, user_id):
        """获取购物车商品种类数"""
        key = f"{self.cart_prefix}{user_id}"
        return self.r.hlen(key)
    def get_total_items(self, user_id):
        """获取购物车商品总数"""
        key = f"{self.cart_prefix}{user_id}"
        cart_items = self.r.hvals(key)
        return sum(int(qty) for qty in cart_items)
# 使用购物车
def shopping_cart_demo():
    cart = ShoppingCart()
    # 添加商品
    cart.add_item("user001", "product_001", 2)
    cart.add_item("user001", "product_002", 1)
    cart.add_item("user001", "product_003", 3)
    # 查看购物车
    cart.get_cart("user001")
    # 更新数量
    cart.update_quantity("user001", "product_001", 5)
    # 移除商品
    cart.remove_item("user001", "product_002")
    # 查看统计
    print(f"\n商品种类: {cart.get_cart_count('user001')}")
    print(f"商品总数: {cart.get_total_items('user001')}")
    # 清空购物车
    cart.clear_cart("user001")
shopping_cart_demo()

性能优化和注意事项

class RedisHashOptimizer:
    def __init__(self):
        self.r = redis.Redis(
            host='localhost',
            port=6379,
            db=0,
            decode_responses=True,
            socket_connect_timeout=5,
            socket_timeout=5,
            retry_on_timeout=True
        )
    def bulk_insert(self, data_dict, batch_size=100):
        """批量插入优化"""
        pipeline = self.r.pipeline()
        for i, (key, value) in enumerate(data_dict.items(), 1):
            pipeline.hset("bulk:data", key, value)
            # 每 batch_size 条执行一次
            if i % batch_size == 0:
                pipeline.execute()
                pipeline = self.r.pipeline()
        # 执行剩余的
        pipeline.execute()
    def memory_optimization(self):
        """内存优化建议"""
        # 使用较短的字段名
        # 而不是 'user_name'，使用 'name'
        # 而不是 'user_email'，使用 'email'
        # 使用哈希而不是字符串存储多个字段
        # 坏做法
        self.r.set("user:1:name", "张三")
        self.r.set("user:1:email", "zhang@example.com")
        # 好做法
        self.r.hset("user:1", mapping={
            "name": "张三",
            "email": "zhang@example.com"
        })
    def expiration_management(self):
        """设置过期时间"""
        # 为整个哈希设置过期
        self.r.hset("session:token123", "user_id", "1001")
        self.r.expire("session:token123", 3600)  # 1小时后过期
        # 或者在设置时直接指定
        self.r.hset("temp:data", "value", "something")
        self.r.expire("temp:data", 300)  # 5分钟后过期
# 运行优化示例
optimizer = RedisHashOptimizer()
optimizer.memory_optimization()
optimizer.expiration_management()

这个案例涵盖了 Python 操作 Redis 哈希的常见场景，包括基本的 CRUD 操作、业务应用（用户管理、购物车）、以及性能优化技巧，实际使用时可以根据具体需求调整。

本文地址： https://www.bmjup.com/post/12253.html