怎样用Python操作Redis？redis-py连接方法

pip install redis

import redis

# 连接到本地Redis服务器，默认端口6379，选择db0
# decode_responses=True 会自动将从Redis获取的字节数据解码为UTF-8字符串
try:
    r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
    print("成功连接到Redis！")

    # 简单的SET和GET操作
    r.set('mykey', 'Hello from Python!')
    value = r.get('mykey')
    print(f"获取到的值: {value}")

    # 判断键是否存在
    if r.exists('mykey'):
        print("'mykey' 存在。")

    # 删除键
    r.delete('mykey')
    if not r.exists('mykey'):
        print("'mykey' 已被删除。")

    # 操作列表 (LPUSH, LRANGE)
    r.lpush('mylist', 'item1', 'item2', 'item3')
    list_items = r.lrange('mylist', 0, -1)
    print(f"列表 'mylist' 中的元素: {list_items}")
    r.delete('mylist') # 清理

    # 操作哈希 (HSET, HGETALL)
    r.hset('myhash', mapping={'name': 'Alice', 'age': 30, 'city': 'New York'})
    hash_data = r.hgetall('myhash')
    print(f"哈希 'myhash' 中的数据: {hash_data}")
    r.delete('myhash') # 清理

except redis.exceptions.ConnectionError as e:
    print(f"连接Redis失败: {e}")
except Exception as e:
    print(f"发生其他错误: {e}")

r = redis.Redis(host='your_redis_host', port=6379, password='your_redis_password', db=0, decode_responses=True)

r = redis.Redis(host='localhost', port=6379, socket_connect_timeout=5, socket_timeout=5, decode_responses=True)

# 假设你需要SSL连接
# r = redis.Redis(
#     host='your_ssl_redis_host',
#     port=6379,
#     ssl=True,
#     ssl_ca_certs='/path/to/ca_certificate.pem',
#     # 如果需要客户端证书认证
#     # ssl_certfile='/path/to/client_certificate.pem',
#     # ssl_keyfile='/path/to/client_key.pem',
#     decode_responses=True
# )

import redis

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 创建一个管道
pipe = r.pipeline()

# 在管道中添加多个命令
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.get('key1')
pipe.delete('key2')
pipe.lpush('mylist_pipe', 'a', 'b', 'c')
pipe.lrange('mylist_pipe', 0, -1)

# 执行管道中的所有命令，并获取结果列表
# 结果的顺序与命令添加的顺序一致
results = pipe.execute()
print(f"Pipelining执行结果: {results}")

# 清理
r.delete('key1', 'mylist_pipe')

# 事务处理
with r.pipeline() as pipe:
    # 开启事务
    pipe.multi()
    # 在事务中添加命令
    pipe.set('transaction_key_1', 'tx_value_1')
    pipe.incr('transaction_counter')
    pipe.lpush('transaction_list', 'item_A', 'item_B')

    # 执行事务，如果期间没有 WATCH 监视的键被修改，则所有命令原子性执行
    try:
        transaction_results = pipe.execute()
        print(f"事务执行结果: {transaction_results}")
    except redis.exceptions.WatchError:
        print("事务执行失败，因为被监视的键发生了变化。")
    except Exception as e:
        print(f"事务执行中发生错误: {e}")

# 清理
r.delete('transaction_key_1', 'transaction_counter', 'transaction_list')

# 乐观锁示例
r.set('balance', 100)

with r.pipeline() as pipe:
    while True:
        try:
            # 监视 'balance' 键
            pipe.watch('balance')
            current_balance = int(pipe.get('balance'))

            if current_balance < 10:
                pipe.unwatch() # 如果余额不足，取消监视并退出
                print("余额不足，无法扣款。")
                break

            # 开启事务
            pipe.multi()
            pipe.decr('balance', 10) # 扣款10

            # 执行事务，如果 'balance' 在 watch 之后被修改，这里会抛出 WatchError
            pipe.execute()
            print(f"成功扣款，当前余额: {r.get('balance')}")
            break # 成功则退出循环

        except redis.exceptions.WatchError:
            print("余额在事务执行前被修改，重试中...")
            # 循环会继续，重新尝试事务
        except Exception as e:
            print(f"发生错误: {e}")
            break

# 清理
r.delete('balance')

import redis
import threading
import time

# 创建一个连接池
# max_connections 参数限制了连接池中最大连接数，防止资源耗尽
pool = redis.ConnectionPool(host='localhost', port=6379, db=0, decode_responses=True, max_connections=10)

def worker(thread_id):
    # 每个线程从连接池中获取一个连接
    # redis-py 的 Redis 客户端实例是线程安全的
    r = redis.Redis(connection_pool=pool)
    try:
        key = f"thread_key_{thread_id}"
        r.set(key, f"Hello from thread {thread_id}!")
        value = r.get(key)
        print(f"Thread {thread_id}: Set and Got '{value}'")
        r.delete(key)
    except Exception as e:
        print(f"Thread {thread_id} encountered error: {e}")
    finally:
        # 连接会自动返回到连接池
        pass # 无需手动释放，因为 connection_pool 管理了

# 启动多个线程来模拟并发访问
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print("\n所有线程任务完成。")