腾讯云分布式缓存数据库 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}')# 设置带过期时间的 Keyr.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 asyncioimport redis.asyncio as aioredisasync 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 客户端接入。

注意：说明：

十、基础连接示例

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, ClusterNodestartup_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 asynciofrom redis.asyncio.cluster import RedisCluster, ClusterNodeasync 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())

十三、连接参数说明

十四、连接超时

·确认客户端所在 CVM 与分布式缓存数据库实例在同一 VPC 内。

·确认安全组规则已放通对应端口。

·确认连接地址和端口正确。

十五、认证失败

·代理模式：默认账号使用 实例 ID:密码，自定义账号使用 用户名@密码。

·直连模式：复用 Redis ACL 认证，默认账号仅传密码（AUTH password），自定义账号需分别传入账号名与密码（AUTH account password），不支持 用户名@密码 拼接格式。

·确认密码中无特殊字符导致解析异常。

十六、直连模式 MOVED 错误

·确认使用 RedisCluster 类而非 Redis 类。

·确认已在控制台开启直连模式。

·确认 startup_nodes 地址为分片节点地址而非 VIP 地址。

十七、连接池耗尽

·检查 max_connections 是否设置合理。

·确认业务代码中没有连接泄漏（未正确释放连接）。

·考虑增大连接池或优化业务并发模型。