腾讯云分布式缓存数据库（兼容 Redis） Python 连接示

本文介绍使用 Python 客户端通过代理模式和直连模式连接腾讯云分布式缓存数据库实例的示例。
准备工作
环境要求
Python 3.7及以上版本。
pip 包管理器。
安装客户端
pip install redis
获取连接信息
1. 登录 分布式缓存数据库控制台。
2. 在实例列表中，单击目标实例 ID，进入实例详情页。
3. 在网络信息区域，获取实例的连接地址和端口号。
代理模式：获取 VIP 地址和端口（默认6379）。
直连模式：获取各分片节点地址和端口（默认6379）。
4. 在配置信息区域，获取实例的连接密码。代理模式与直连模式的认证方式不同：
代理模式：默认账号使用 实例 ID:密码 格式，自定义账号使用 用户名@密码 格式；免密认证实例无需密码。
直连模式：复用 Redis 原生 ACL 认证机制，不支持 用户名@密码 拼接格式，支持以下两种认证方式：
默认账号：仅需密码，即 AUTH password。
自定义账号：需同时指定账号名与密码，即 AUTH account password。
代理模式连接示例
代理模式下，客户端通过 Proxy 代理层提供的统一 VIP 地址连接实例，Proxy 自动处理请求路由和故障转移，客户端无需感知后端拓扑变化。
连接池示例（推荐）
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',   # 代理模式自定义账号：用户名@密码整体作为密码传入（区别于直连模式 ACL 的账号名/密码分离传参）
    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'))
SSL 加密连接示例
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'))
异步连接示例（asyncio）
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())
直连模式连接示例
直连模式下，客户端使用 Redis Cluster 协议直接连接各分片节点，本地处理 MOVED/ASK 重定向，可直接使用社区标准的 Redis Cluster 客户端接入。
说明：
直连模式仅适用于集群架构实例。
需使用支持 Redis Cluster 协议的客户端（redis-py 的 RedisCluster 类）。
需在控制台开启直连模式并获取各分片节点的连接地址。
认证方式复用 Redis 原生 ACL 机制：默认账号仅传密码，自定义账号需分别传入账号名与密码，不支持 用户名@密码 拼接格式。
直连模式暂不支持 SSL 加密连接。
基础连接示例
from redis.cluster import RedisCluster, ClusterNode
﻿
# 直连模式下各分片节点地址
startup_nodes = [
    ClusterNode('192.xx.xx.2', 6379),
    ClusterNode('192.xx.xx.3', 6379),
    ClusterNode('192.xx.xx.4', 6379)
]
﻿
# 创建集群客户端
rc = RedisCluster(
    startup_nodes=startup_nodes,
    password='password',            # 直连模式默认账号仅传密码，不带实例 ID 前缀
    socket_connect_timeout=2,       # 连接超时（秒）
    socket_timeout=2,               # 读写超时（秒）
    decode_responses=True,
    max_connections=200,            # 每个节点最大连接数
    retry_on_timeout=True           # 超时自动重试
)
﻿
# 写入数据（客户端自动路由到正确的分片）
rc.set('cluster:key1', 'value1')
rc.set('cluster:key2', 'value2')
rc.set('cluster:key3', 'value3')
﻿
# 读取数据
print(f'key1: {rc.get("cluster:key1")}')
print(f'key2: {rc.get("cluster:key2")}')
print(f'key3: {rc.get("cluster:key3")}')
﻿
# 哈希操作
rc.hset('user:{1001}', mapping={'name': '张三', 'age': '28'})
user_info = rc.hgetall('user:{1001}')
print(f'user info: {user_info}')
﻿
# 使用 Hash Tag 确保相关 Key 路由到同一分片
rc.set('{order:1001}:status', 'paid')
rc.set('{order:1001}:amount', '99.00')
rc.set('{order:1001}:time', '2025-01-01 10:00:00')
print(f'order status: {rc.get("{order:1001}:status")}')
﻿
# Pipeline 操作（同一 Hash Tag 的 Key）
pipe = rc.pipeline()
pipe.set('{batch}:k1', 'v1')
pipe.set('{batch}:k2', 'v2')
pipe.set('{batch}:k3', 'v3')
pipe.get('{batch}:k1')
pipe.get('{batch}:k2')
pipe.get('{batch}:k3')
results = pipe.execute()
print(f'pipeline results: {results}')
﻿
# 关闭连接
rc.close()
自定义账号（ACL）连接示例
直连模式复用 Redis 原生 ACL 认证机制，使用自定义账号时需分别传入账号名与密码，不支持 用户名@密码 拼接格式。
from redis.cluster import RedisCluster, ClusterNode
﻿
startup_nodes = [
    ClusterNode('192.xx.xx.2', 6379),
    ClusterNode('192.xx.xx.3', 6379),
    ClusterNode('192.xx.xx.4', 6379)
]
﻿
# ===== ACL 认证机制：账号名与密码是两个独立字段 =====
# 直连模式的多账号能力基于 Redis 原生 ACL：控制台创建的每个自定义账号
# 都是一个独立的 ACL user，认证时客户端向节点发送 AUTH <account> <password>。
# 因此需将账号名与密码分别传入（等价于执行 AUTH account password），
# 不要写成代理模式的 "myuser@MyPassword" 拼接格式。
rc = RedisCluster(
    startup_nodes=startup_nodes,
    username='myuser',              # ACL 账号名 → AUTH account
    password='MyPassword',          # 账号密码   → AUTH password（纯密码，无需拼接）
    socket_connect_timeout=2,
    socket_timeout=2,
    decode_responses=True,
    max_connections=200
)
﻿
rc.set('key', 'value')
print(rc.get('key'))
﻿
rc.close()
异步集群连接示例
import asyncio
from redis.asyncio.cluster import RedisCluster, ClusterNode
﻿
async def main():
    startup_nodes = [
        ClusterNode('192.xx.xx.2', 6379),
        ClusterNode('192.xx.xx.3', 6379),
        ClusterNode('192.xx.xx.4', 6379)
    ]
﻿
    rc = RedisCluster(
        startup_nodes=startup_nodes,
        password='password',            # 直连模式默认账号仅传密码，不带实例 ID 前缀
        socket_connect_timeout=2,
        socket_timeout=2,
        decode_responses=True,
        max_connections=200
    )
﻿
    await rc.set('async_cluster_key', 'async_value')
    value = await rc.get('async_cluster_key')
    print(f'async cluster get: {value}')
﻿
    await rc.aclose()
﻿
asyncio.run(main())
连接参数说明
参数
说明
推荐值
host
实例连接地址。代理模式为 VIP 地址，直连模式为分片节点地址
-
port
实例端口号
6379
password
认证密码
默认账号格式：实例 ID:密码
max_connections
最大连接数
200
socket_connect_timeout
连接超时（秒）
2
socket_timeout
读写超时（秒）
2
health_check_interval
健康检查间隔（秒）
30
decode_responses
自动解码响应为字符串
True
retry_on_timeout
超时自动重试
True
常见问题
连接超时
确认客户端所在 CVM 与分布式缓存数据库实例在同一 VPC 内。
确认安全组规则已放通对应端口。
确认连接地址和端口正确。
认证失败
代理模式：默认账号使用 实例 ID:密码，自定义账号使用 用户名@密码。
直连模式：复用 Redis ACL 认证，默认账号仅传密码（AUTH password），自定义账号需分别传入账号名与密码（AUTH account password），不支持 用户名@密码 拼接格式。
确认密码中无特殊字符导致解析异常。
直连模式 MOVED 错误
确认使用 RedisCluster 类而非 Redis 类。
确认已在控制台开启直连模式。
确认 startup_nodes 地址为分片节点地址而非 VIP 地址。
连接池耗尽
检查 max_connections 是否设置合理。
确认业务代码中没有连接泄漏（未正确释放连接）。
考虑增大连接池或优化业务并发模型。
Python 连接示例

本页目录：

准备工作

环境要求

安装客户端

获取连接信息

代理模式连接示例

连接池示例（推荐）

自定义账号连接示例

SSL 加密连接示例

异步连接示例（asyncio）

直连模式连接示例

基础连接示例

自定义账号（ACL）连接示例

异步集群连接示例

连接参数说明

常见问题

连接超时

认证失败

直连模式 MOVED 错误

连接池耗尽

参数	说明	推荐值
host	实例连接地址。代理模式为 VIP 地址，直连模式为分片节点地址	-
port	实例端口号	6379
password	认证密码	默认账号格式：`实例 ID:密码`
max_connections	最大连接数	200
socket_connect_timeout	连接超时（秒）	2
socket_timeout	读写超时（秒）	2
health_check_interval	健康检查间隔（秒）	30
decode_responses	自动解码响应为字符串	True
retry_on_timeout	超时自动重试	True