Python 连接示例

pip install redis

import redis

# 创建连接池
pool = redis.ConnectionPool(
    host='192.xx.xx.2',            # 实例 VIP 地址
    port=6379,                      # 实例端口
    password='crs-xxxx:密码',       # 默认账号密码格式：实例ID:密码
    db=0,                           # 数据库编号
    max_connections=200,            # 最大连接数
    socket_connect_timeout=2,       # 连接超时（秒）
    socket_timeout=2,               # 读写超时（秒）
    decode_responses=True,          # 自动解码为字符串
    health_check_interval=30        # 健康检查间隔（秒）
)

# 创建 Redis 客户端
r = redis.Redis(connection_pool=pool)

# 写入数据
r.set('name', '腾讯云分布式缓存数据库')

# 读取数据
value = r.get('name')
print(f'get name: {value}')

# 设置带过期时间的 Key
r.setex('session:user123', 3600, 'session_data')

# 哈希操作
r.hset('user:1001', mapping={'name': '张三', 'age': '28'})
user_info = r.hgetall('user:1001')
print(f'user info: {user_info}')

# 列表操作
r.lpush('queue:tasks', 'task1', 'task2', 'task3')
task = r.rpop('queue:tasks')
print(f'dequeued task: {task}')

# Pipeline 批量操作
pipe = r.pipeline()
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.set('key3', 'value3')
pipe.get('key1')
pipe.get('key2')
pipe.get('key3')
results = pipe.execute()
print(f'pipeline results: {results}')

import redis

# 自定义账号密码格式：用户名@密码
pool = redis.ConnectionPool(
    host='192.xx.xx.2',
    port=6379,
    password='myuser@MyPassword',   # 代理模式自定义账号：用户名@密码整体作为密码传入
    db=0,
    max_connections=200,
    socket_connect_timeout=2,
    socket_timeout=2,
    decode_responses=True
)

r = redis.Redis(connection_pool=pool)
r.set('key', 'value')
print(r.get('key'))

import redis

# SSL 加密连接
pool = redis.ConnectionPool(
    host='192.xx.xx.2',
    port=6379,
    password='crs-xxxx:密码',
    db=0,
    max_connections=200,
    socket_connect_timeout=2,
    socket_timeout=2,
    decode_responses=True,
    ssl=True,                       # 启用 SSL
    ssl_ca_certs='/path/to/ca.pem', # CA 证书路径
    ssl_cert_reqs='required'        # 验证服务端证书
)

r = redis.Redis(connection_pool=pool)
r.set('ssl_test', 'encrypted_connection')
print(r.get('ssl_test'))

import asyncio
import redis.asyncio as aioredis

async def main():
    # 创建异步连接池
    pool = aioredis.ConnectionPool(
        host='192.xx.xx.2',
        port=6379,
        password='crs-xxxx:密码',
        db=0,
        max_connections=200,
        socket_connect_timeout=2,
        socket_timeout=2,
        decode_responses=True
    )

    r = aioredis.Redis(connection_pool=pool)

    # 异步写入
    await r.set('async_key', 'async_value')

    # 异步读取
    value = await r.get('async_key')
    print(f'async get: {value}')

    # 异步 Pipeline
    pipe = r.pipeline()
    pipe.set('key1', 'value1')
    pipe.set('key2', 'value2')
    pipe.get('key1')
    pipe.get('key2')
    results = await pipe.execute()
    print(f'async pipeline: {results}')

    # 关闭连接
    await r.aclose()
    await pool.aclose()

asyncio.run(main())