云数据库Tair（兼容 Redis）与RDS MySQL协同工作_云数据库 RDS(RDS)-阿里云帮助中心

当您使用云数据库 RDS MySQL 并需要提升数据处理速率和降低访问延迟时，阿里云为您提供了缓存数据持久化的高效方案，该解决方案通过整合云数据库 Tair（兼容 Redis）与 RDS、云数据库 Memcache 与 RDS 协同工作，确保数据的高速访问和持久化存储，具备高吞吐、低延迟等优势。

背景信息

与 RDS 相比，云数据库缓存产品有如下两个特性：

响应速度快，云数据库 Tair（兼容 Redis）和云数据库 Memcache 请求的时延通常在几毫秒以内。
缓存区能够提供比 RDS 更高的 QPS（每秒处理请求数）。

操作步骤

RDS MySQL 与云数据库 Tair（兼容 Redis）搭配

前提条件

已创建云服务器 ECS、云数据库 Tair（兼容 Redis）和云数据库 RDS MySQL。
- 如果仅需要通过内网访问，则建议将三者创建在同一 VPC 下。
- 如果已创建相关云服务，但三者不在同一 VPC 下，则可以分别为云数据库 Tair（兼容 Redis）和云数据库 RDS MySQL 开通外网连接地址，通过外网访问。

import json
import redis
import pymysql
# 定义MySQL连接参数
mysql_host = '<RDS MySQL连接地址>'
mysql_user = '<用户名>'
mysql_password = '<密码>'
mysql_port = 3306
mysql_charset = 'utf8'
# 定义Tair（兼容 Redis）连接参数
redis_host = '<Tair连接地址>'
redis_port = 6379
# Tair（兼容 Redis）密码格式（账号:密码），如果没有密码则为空字符串
redis_password = '<Tair密码>'
redis_db = 0
def create_database_and_tables():
    db = pymysql.connect(host=mysql_host,
                         user=mysql_user,
                         password=mysql_password,
                         port=mysql_port,
                         charset=mysql_charset)
    cursor = db.cursor()
    # 创建测试数据库
    cursor.execute("CREATE DATABASE IF NOT EXISTS testdb;")
    # 选择数据库
    cursor.execute("USE testdb;")
    # 创建测试表
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS student (
        s_id INT AUTO_INCREMENT PRIMARY KEY,
        s_name VARCHAR(255) NOT NULL
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
    # 插入测试数据
    cursor.execute("""
    INSERT INTO student (s_name) VALUES
    ('Zhangsan'),
    ('Lisi'),
    ('Wangwu')
    ON DUPLICATE KEY UPDATE s_name = VALUES(s_name);
    db.commit()
    cursor.close()
    db.close()
def fetch_from_mysql():
    db = pymysql.connect(host=mysql_host,
                         user=mysql_user,
                         password=mysql_password,
                         database="testdb",
                         port=mysql_port,
                         charset=mysql_charset)
    cursor = db.cursor()
    cursor.execute("SELECT * FROM student")
    rows = cursor.fetchall()
    cursor.close()
    db.close()
    return rows
def cache_to_redis(redis_client, key, data):
    # 将数据编码成JSON字符串以存储复杂数据类型
    json_data = json.dumps(data)
    # 存储到Tair（兼容 Redis），并设置有效期为600秒（10分钟）
    redis_client.setex(key, 600, json_data)
def get_from_redis(redis_client, key):
    # 尝试从Tair（兼容 Redis）获取数据
    json_data = redis_client.get(key)
    if json_data:
        # 如果有数据，进行JSON解码
        data = json.loads(json_data)
        return data
    else:
        return None
def main():
    # 创建Tair（兼容 Redis）客户端
    redis_client = redis.StrictRedis(
        host=redis_host,
        port=redis_port,
        password=redis_password,
        db=redis_db,
        decode_responses=True  # 自动解码响应
    # 创建MySQL测试表并插入测试数据
    create_database_and_tables()
    # 定义Tair（兼容 Redis）中存储学生信息的键
    redis_key = 'students'
    # 尝试从Tair（兼容 Redis）获取数据
    students = get_from_redis(redis_client, redis_key)
    if students:
        print("从Tair查询到数据：")
        print(students)
    else:
        print("Tair中未查询到数据，从RDS MySQL查询到数据：")
        # 从RDS MySQL获取数据
        students = fetch_from_mysql()
        if students:
            print(students)
            # 缓存数据到Tair（兼容 Redis）
            cache_to_redis(redis_client, redis_key, students)
if __name__ == '__main__':
    main()

```
python3 test.py
```
- ```
Tair中未查询到数据，从RDS MySQL查询到数据：
((1, 'Zhangsan'), (2, 'Lisi'), (3, 'Wangwu'))
```
- ```
从Tair查询到数据：
[[1, 'Zhangsan'], [2, 'Lisi'], [3, 'Wangwu']]
```

import json
import pymysql
import memcache
# 定义MySQL连接参数
mysql_host = '<RDS MySQL连接地址>'
mysql_user = '<用户名>'
mysql_password = '<密码>'
mysql_port = 3306
mysql_charset = 'utf8'
# 定义Memcache连接参数
memcache_host = '<Memcache连接地址>:<端口>'
def create_database_and_tables():
    db = pymysql.connect(host=mysql_host,
                         user=mysql_user,
                         password=mysql_password,
                         port=mysql_port,
                         charset=mysql_charset)
    cursor = db.cursor()
    # 创建测试数据库
    cursor.execute("CREATE DATABASE IF NOT EXISTS testdb;")
    # 选择数据库
    cursor.execute("USE testdb;")
    # 创建测试表
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS student (
        s_id INT AUTO_INCREMENT PRIMARY KEY,
        s_name VARCHAR(255) NOT NULL
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
    # 插入测试数据
    cursor.execute("""
    INSERT INTO student (s_name) VALUES
    ('Zhangsan'),
    ('Lisi'),
    ('Wangwu')
    ON DUPLICATE KEY UPDATE s_name = VALUES(s_name);
    db.commit()
    cursor.close()
    db.close()
def fetch_from_mysql():
    db = pymysql.connect(host=mysql_host,
                         user=mysql_user,
                         password=mysql_password,
                         database="testdb",
                         port=mysql_port,
                         charset=mysql_charset)
    cursor = db.cursor()
    cursor.execute("SELECT * FROM student")
    rows = cursor.fetchall()
    cursor.close()
    db.close()
    return rows
def cache_to_memcache(memcache_client, key, data):
    # 将数据编码成JSON字符串以存储复杂数据类型
    json_data = json.dumps(data)
    # 存储到Memcache，并设置有效期为600秒（10分钟）
    memcache_client.set(key, json_data, time=600)
def get_from_memcache(memcache_client, key):
    # 尝试从Memcache获取数据
    json_data = memcache_client.get(key)
    if json_data:
        # 如果有数据，进行JSON解码
        data = json.loads(json_data)
        return data
    else:
        return None
def main():
    # 创建Memcache客户端
    memcache_client = memcache.Client([memcache_host], debug=0)
    # 创建MySQL测试表并插入测试数据
    create_database_and_tables()
    # 定义Memcache中存储学生信息的键
    memcache_key = 'students'
    # 尝试从Memcache获取数据
    students = get_from_memcache(memcache_client, memcache_key)
    if students:
        print("从Memcache查询到数据：")
        print(students)
    else:
        print("Memcache中未查询到数据，从RDS MySQL查询到数据：")
        # 从RDS MySQL获取数据
        students = fetch_from_mysql()
        if students:
            print(students)
            # 缓存数据到Memcache
            cache_to_memcache(memcache_client, memcache_key, students)
if __name__ == '__main__':
    main()

```
python3 test.py
```
- ```
Memcache中未查询到数据，从RDS MySQL查询到数据：
((1, 'Zhangsan'), (2, 'Lisi'), (3, 'Wangwu'))
```
- ```
从Memcache查询到数据：
[[1, 'Zhangsan'], [2, 'Lisi'], [3, 'Wangwu']]
```

缓存数据持久化

背景信息

操作步骤

RDS MySQL 与云数据库 Tair（兼容 Redis）搭配

前提条件

配置步骤

RDS MySQL 与云数据库 Memcache 搭配

前提条件

配置步骤

相关文档