AI 助理
备案控制台
文档
输入文档关键字查找
首页云数据库 RDSRDS MySQL数据库操作指南附录缓存数据持久化

缓存数据持久化

更新时间：2024-11-19 06:16:12
产品详情
我的收藏
重要

本文中含有需要您注意的重要提示信息，忽略该信息可能对您的业务造成影响，请务必仔细阅读。

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

背景信息

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

  • 响应速度快，云数据库 Tair（兼容 Redis）和云数据库Memcache请求的时延通常在几毫秒以内。

  • 缓存区能够提供比RDS更高的QPS（每秒处理请求数）。

操作步骤

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

前提条件

  • 已创建云服务器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创建账号。

配置步骤

  1. 登录云服务器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()

  2. 运行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']]

前提条件

  • 已创建云服务器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已开启免密登录。更多信息，请参见免密码访问。

配置步骤

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

    警告

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

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

  2. 运行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访问KMS下一篇：RDS与智能开放搜索OpenSearch
  • 本页导读 （1）
  • 背景信息
  • 操作步骤
  • 相关文档