云数据库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开通外网连接地址，通过外网访问。
说明
- 本文示例的云服务器ECS镜像版本为：Alibaba Cloud Linux 3.2104 LTS 64位。
- 本文示例使用的开发语言为Python，请提前在云服务器ECS中安装Python 3、pip 3、pymysql和redis-py。
  pymysql和redis-py安装命令如下：
```
sudo pip3 install pymysql
sudo pip3 install redis
```
白名单设置：
- 如果使用内网，请将VPC网段加入到云数据库 Tair（兼容 Redis）和云数据库RDS MySQL的白名单中。更多信息，请参见Tair设置白名单、RDS MySQL设置IP白名单。
- 如果使用外网，请分别为云数据库 Tair（兼容 Redis）和云数据库RDS MySQL开通外网连接地址，并将ECS的公网IP加入到云数据库 Tair（兼容 Redis）和云数据库RDS MySQL的白名单中。更多信息，请参见Tair申请公网连接地址、RDS MySQL申请外网地址。
云数据库 Tair（兼容 Redis）和云数据库RDS MySQL已创建数据库账号和密码。更多信息，请参见Tair创建与管理账号、RDS MySQL创建账号。

配置步骤

登录云服务器ECS，编写Python脚本（本文示例命名为test.py），模拟业务场景，当从云数据库 Tair（兼容 Redis）缓存中查询不到结果时，从云数据库RDS MySQL中查询。

警告

本示例所展示的代码仅用于演示配置方式，请勿在实际业务代码中将user和password以明文方式设置在代码中，建议使用外部配置文件或环境变量等其他方式进行处理后，再在代码中引用。

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()

运行test.py。
```
python3 test.py
```
- 首次运行，由于Tair缓存中没有数据，从RDS MySQL中读取，返回结果示例：
```
Tair中未查询到数据，从RDS MySQL查询到数据：
((1, 'Zhangsan'), (2, 'Lisi'), (3, 'Wangwu'))
```
- 再次运行时，由于第一次查询后已将查询数据缓存至Tair中，因此第二次直接从Tair缓存中读取数据。返回结果示例：
```
从Tair查询到数据：
[[1, 'Zhangsan'], [2, 'Lisi'], [3, 'Wangwu']]
```

RDS MySQL与云数据库Memcache搭配

前提条件

已创建云服务器ECS、云数据库Memcache和云数据库RDS MySQL，且确保三者在同一VPC下。
说明
- 云数据库Memcache暂不提供外网访问，仅支持通过内网访问，因此，请确保三者在同一VPC下。
- 本文示例的云服务器ECS镜像版本为：Alibaba Cloud Linux 3.2104 LTS 64位。
- 本文示例使用的开发语言为Python，请提前在云服务器ECS中安装Python 3、pip 3、pymysql和python-memcached。
  pymysql和python-memcached安装命令如下：
```
sudo pip3 install pymysql
sudo pip3 install python-memcached
```
已将VPC网段加入到云数据库Memcache和云数据库RDS MySQL的白名单中。更多信息，请参见RDS MySQL设置IP白名单、Memcache设置IP白名单。
云数据库RDS MySQL已创建数据库账号和密码。更多信息，请参见创建账号。
云数据库Memcache已开启免密登录。更多信息，请参见免密码访问。

配置步骤

登录云服务器ECS，编写Python脚本（本文示例命名为test.py），模拟业务场景，当从云数据库Memcache缓存中查询不到结果时，从云数据库RDS MySQL中查询。

警告

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()

运行test.py。
```
python3 test.py
```
- 首次运行，由于Memcache缓存中没有数据，从RDS MySQL中读取，返回结果示例：
```
Memcache中未查询到数据，从RDS MySQL查询到数据：
((1, 'Zhangsan'), (2, 'Lisi'), (3, 'Wangwu'))
```
- 再次运行时，由于第一次查询后已将查询数据缓存至Memcache中，因此第二次直接从Memcache缓存中读取数据。返回结果示例：
```
从Memcache查询到数据：
[[1, 'Zhangsan'], [2, 'Lisi'], [3, 'Wangwu']]
```

背景信息

操作步骤

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

前提条件

配置步骤

RDS MySQL与云数据库Memcache搭配

前提条件

配置步骤

相关文档