文档

云数据库Redis与RDS MySQL协同工作

更新时间:
重要

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

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

背景信息

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

  • 响应速度快,云数据库Redis和云数据库Memcache请求的时延通常在几毫秒以内。

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

操作步骤

RDS MySQL与云数据库Redis搭配

前提条件

  • 已创建云服务器ECS、云数据库Redis和云数据库RDS MySQL。

    • 如果仅需要通过内网访问,则建议将三者创建在同一VPC下。

    • 如果已创建相关云服务,但三者不在同一VPC下,则可以分别为云数据库Redis和云数据库RDS MySQL开通外网连接地址,通过外网访问。

    说明
    • 本文示例的云服务器ECS镜像版本为:Alibaba Cloud Linux 3.2104 LTS 64位

    • 本文示例使用的开发语言为Python,请提前在云服务器ECS中安装Python 3pip 3pymysqlredis-py

      pymysqlredis-py安装命令如下:

      sudo pip3 install pymysql
      sudo pip3 install redis
  • 白名单设置:

  • 云数据库Redis和云数据库RDS MySQL已创建数据库账号和密码。更多信息,请参见Redis创建与管理账号RDS MySQL创建账号

配置步骤

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

    警告

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

    import json
    import redis
    import pymysql
    
    # 定义MySQL连接参数
    mysql_host = '<RDS MySQL连接地址>'
    mysql_user = '<用户名>'
    mysql_password = '<密码>'
    mysql_port = 3306
    mysql_charset = 'utf8'
    
    # 定义Redis连接参数
    redis_host = '<Redis连接地址>'
    redis_port = 6379
    # Redis密码格式(账号:密码),如果没有密码则为空字符串
    redis_password = '<Redis密码>'
    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)
        # 存储到Redis,并设置有效期为600秒(10分钟)
        redis_client.setex(key, 600, json_data)
    
    
    def get_from_redis(redis_client, key):
        # 尝试从Redis获取数据
        json_data = redis_client.get(key)
        if json_data:
            # 如果有数据,进行JSON解码
            data = json.loads(json_data)
            return data
        else:
            return None
    
    
    def main():
        # 创建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()
    
        # 定义Redis中存储学生信息的键
        redis_key = 'students'
    
        # 尝试从Redis获取数据
        students = get_from_redis(redis_client, redis_key)
    
        if students:
            print("从Redis查询到数据:")
            print(students)
        else:
            print("Redis中未查询到数据,从RDS MySQL查询到数据:")
            # 从RDS MySQL获取数据
            students = fetch_from_mysql()
            if students:
                print(students)
                # 缓存数据到Redis
                cache_to_redis(redis_client, redis_key, students)
    
    
    if __name__ == '__main__':
        main()
    
  2. 运行test.py

    python3 test.py
    • 首次运行,由于Redis缓存中没有数据,从RDS MySQL中读取,返回结果示例:

      Redis中未查询到数据,从RDS MySQL查询到数据:
      ((1, 'Zhangsan'), (2, 'Lisi'), (3, 'Wangwu'))
    • 再次运行时,由于第一次查询后已将查询数据缓存至Redis中,因此第二次直接从Redis缓存中读取数据。返回结果示例:

      从Redis查询到数据:
      [[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 3pip 3pymysqlpython-memcached

      pymysqlpython-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中查询。

    警告

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

    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']]

相关文档

  • 本页导读 (1)
文档反馈