当前位置:   article > 正文

条条大路通罗马 —— 使用 Go-Redis 连接 Amazon ElastiCache for Redis 集群

go redis maxretries 选项

2029a98f5b0a2cf270fc7e0902d0ac1a.gif

一、前言

Amazon ElastiCache 是一种 Web 服务,可让用户在云中轻松设置、管理和扩展分布式内存数据存储或缓存环境。它可以提供高性能、可扩展且具有成本效益的缓存解决方案。同时,它可以帮助消除与部署和管理分布式缓存环境相关的复杂性。

Amaozn ElastiCache for Redis 集群是一个或多个缓存节点的集合,其中所有节点都运行 Redis 缓存引擎软件的实例。ElastiCache for Redis 启用集群模式比之禁用集群模式拥有更好的可扩展性,尤其是写入可扩展性,更强的高可用性以及更高的资源上限,因而现在越来越多的客户选择 ElastiCache for Redis 启用集群模式。要使用 ElastiCache for Redis 集群(启用集群模式),您需要使用可以支持 Redis 集群模式的客户端。

当您使用 Go 程序连接 ElastiCache 集群时,目前目前主流的 SDK 是 Go-Redis 项目,本篇 Blog 将为您介绍如何使用 Go-Redis 连接和使用 ElastiCache for Redis 集群。 

除此以外,我们还推出了一系列博客,展示了如何在不同语言中,使用不同的支持 ElastiCache 集群模式的客户端对 ElastiCache 集群进行连接和操作,欢迎大家阅读。

二、Go-Redis 测试环境搭建

Go-Redis 是目前排名最高的 Go 语言版本的 Redis client,支持连接哨兵和集群模式的 Redis,并且提供了高级的 API 封装,区别于另一个比较常用的 Go 语言 Redis client 库:Redigo,在服务集成过程中提供更多的功能支持,并且保障 Redis 数据类型安全支持。可以参考 Go-Redis 和 Redigo 对比 去了解更多的差异。

Go-Redis:

https://redis.uptrace.dev/

Go-Redis 和 Redigo 对比:

https://redis.uptrace.dev/guide/Go-Redis-vs-redigo.html

▌ElastiCache for Redis 集群搭建

在亚马逊云科技上搭建 ElastiCache for Redis 集群,可以参考本篇的系列 Blog,条条大路通罗马 —— 使用 Redisson 连接 Amazon ElastiCache for Redis 集群 的2.1章节,这里就不再赘述。(Redis Cluster,打开 Auth+TLS 模式)。

条条大路通罗马 —— 使用 Redisson 连接 Amazon ElastiCache for Redis 集群:

https://aws.amazon.com/cn/blogs/china/connecting-amazon-elasticache-for-redis-cluster-using-redisson/

构建 Golang SDK 测试代码工程的目录结构

  1. [ec2-user src]$ tree test-redis-sdk/
  2. test-redis-sdk/
  3. |-- cmd
  4. |   `-- test-redis
  5. |       `-- redis_test.go
  6. |-- go.mod
  7. `-- go.sum

使用 Go-Redis 最新的版本分支 v8 版本 ,下图总结了初始化 cluster client 常用参数,PoolSize 和 MinIdleConns 控制请求的连接池。Go-Redis 支持 TLS 连接,本示例主要演示打开 TLS+Password 的 ElastiCache for Redis 集群如何接入,如下图所示,参数 Password 已设置,参数 TLSConfig 控制 TLS 开关已打开。如果您连接的 ElastiCache for Redis 集群没有开启 TLS 接入,只需要把 Password 参数置为空字符串,去除 TLSConfig 配置即可(TLSConfig 默认关闭)。

7d74f5abcfe0e3c79c492fd453f66604.png

▌使用 Go-Redis SDK 初始化 cluster client,包括读请求的测试逻辑(源代码)

  1. [ec2-user test-redis]$ cat redis_test.go
  2. package main
  3. import (
  4.    "context"
  5.    "crypto/tls"
  6.    "fmt"
  7.    goredis "github.com/Go-Redis/redis/v8"
  8.    "strconv"
  9.    "sync"
  10.    "testing"
  11.    "time"
  12. )
  13. func TestGoRedisCluster(t *testing.T) {
  14.    var ctx = context.Background()
  15.    rdb := goredis.NewClusterClient(&goredis.ClusterOptions{
  16. Addrs: []string{"cluster-configuration-endpoint:6379"},
  17. Password: "password",     //密码
  18.        //连接池容量及闲置连接数量
  19.        PoolSize:     10, // 连接池最大socket连接数,默认为4倍CPU数, 4 * runtime.NumCPU
  20.        MinIdleConns: 10, //在启动阶段创建指定数量的Idle连接,并长期维持idle状态的连接数不少于指定数量;。
  21.        //超时
  22.        DialTimeout:  5 * time.Second, //连接建立超时时间,默认5秒。
  23.        ReadTimeout:  3 * time.Second, //读超时,默认3秒, -1表示取消读超时
  24.        WriteTimeout: 3 * time.Second, //写超时,默认等于读超时
  25.        PoolTimeout:  4 * time.Second, //当所有连接都处在繁忙状态时,客户端等待可用连接的最大等待时长,默认为读超时+1秒。
  26.        //闲置连接检查包括IdleTimeout,MaxConnAge
  27.        IdleCheckFrequency: 60 * time.Second, //闲置连接检查的周期,默认为1分钟,-1表示不做周期性检查,只在客户端获取连接时对闲置连接进行处理。
  28.        IdleTimeout:        5 * time.Minute,  //闲置超时,默认5分钟,-1表示取消闲置超时检查
  29.        MaxConnAge:         0 * time.Second,  //连接存活时长,从创建开始计时,超过指定时长则关闭连接,默认为0,即不关闭存活时长较长的连接
  30.        //命令执行失败时的重试策略
  31.        MaxRetries:      10,                     // 命令执行失败时,最多重试多少次,默认为0即不重试
  32.        MinRetryBackoff: 8 * time.Millisecond,   //每次计算重试间隔时间的下限,默认8毫秒,-1表示取消间隔
  33.        MaxRetryBackoff: 512 * time.Millisecond, //每次计算重试间隔时间的上限,默认512毫秒,-1表示取消间隔
  34.        TLSConfig: &tls.Config{
  35.            InsecureSkipVerify: true,
  36.        },
  37.        // ReadOnly = true,只择 Slave Node
  38.        // ReadOnly = true 且 RouteByLatency = true 将从 slot 对应的 Master Node 和 Slave Node, 择策略为: 选择PING延迟最低的点
  39.        // ReadOnly = true 且 RouteRandomly = true 将从 slot 对应的 Master Node 和 Slave Node 选择,选择策略为: 随机选择
  40.        ReadOnly: true,
  41.        RouteRandomly: true,
  42.        RouteByLatency: true,
  43.    })
  44.    defer rdb.Close()
  45.    rdb.Set(ctx, "test-0", "value-0", 0)
  46.    rdb.Set(ctx, "test-1", "value-1", 0)
  47.    rdb.Set(ctx, "test-2", "value-2", 0)
  48.    AllMaxRun := 6
  49.    wg := sync.WaitGroup{}
  50.    wg.Add(AllMaxRun)
  51.    for i := 0; i < AllMaxRun; i ++ {
  52.        go func(wg *sync.WaitGroup, idx int) {
  53.            defer wg.Done()
  54.            for i := 0; i < 50000; i++ {
  55.                key := "test-" + strconv.Itoa(i % 3)
  56.                val, err := rdb.Get(ctx, key).Result()
  57.                if err == goredis.Nil {
  58.                    fmt.Println("job-" + strconv.Itoa(idx) + " " + key + " does not exist")
  59.                } else if err != nil {
  60.                    fmt.Printf("err : %s", err.Error())
  61.                } else {
  62.                    fmt.Printf("%s Job-%d %s = %s-%d \n", time.Now().Format("2006-01-02 15:04:05"), idx, key, val, i)
  63.                }
  64.                time.Sleep(500 * time.Millisecond)
  65.            }
  66.        }(&wg, i)
  67.    }
  68.    wg.Wait()
  69.    stats := rdb.PoolStats()
  70.    fmt.Printf("Hits=%d Misses=%d Timeouts=%d TotalConns=%d IdleConns=%d StaleConns=%d\n",
  71.        stats.Hits, stats.Misses, stats.Timeouts, stats.TotalConns, stats.IdleConns, stats.StaleConns)
  72. }

左滑查看更多

三、Go-Redis 读写分离控制测试

Redis cluster 是有 Master 和 Slave 节点,Go-Redis 支持对 Slave 节点的访问,通过配置 ReadOnly 参数,控制 Master 和 Slave 节点的读写管理。下面我们通过不同的配置去做测试验证。

▌ReadOnly 配置规则说明

  1. // 默认false,即只能在主节点上执行读写操作,如果置为true则允许在从节点上执行只含读操作的命令
  2. ReadOnly: false,
  3. // 默认false。 置为true则ReadOnly自动置为true,表示在处理只读命令时,可以在一个slot对应的主节点和所有从节点中选取Ping()的响应时长最短的一个节点来读数据
  4. RouteByLatency: false,
  5. // 默认false。置为true则ReadOnly自动置为true,表示在处理只读命令时,可以在一个slot对应的主节点和所有从节点中随机挑选一个节点来读数据
  6. RouteRandomly: false,

左滑查看更多

▌关闭 ReadOnly 配置测试

修改测试代码,关闭 ReadOnly 配置(三个 ReadOnly 参数配置都置为 false),观察监控仍然是维持10个  conn,但是按照配置说明,服务不会从读节点读取数据。

  1. [ec2-user ~]$ netstat -a | grep 6379 | grep ESTABLISHED | awk '{print $5}' | sort | uniq -c | sort -rn
  2.     10 ip-172-31-18-215.a:6379
  3.     10 ip-172-31-46-118.a:6379
  4.     10 ip-172-31-34-217.a:6379
  5.     10 ip-172-31-31-193.a:6379
  6.     10 ip-172-31-15-157.a:6379
  7.     10 ip-172-31-10-163.a:6379

左滑查看更多

观察对应的连接数,连接数仍然保持在10个。

2ca345b37a3912dde40ee81b850d42b3.png

调整测试代码,增大查询压力,观察 GetTypeCmds 监控指标,可以看到只有 master 节点上是所有的读请求,判断所有的读压力都是分布在所有的 master 节点上。

1e4ff8755fda0d24bfa0bad6e11c9123.png

▌打开 ReadOnly 配置测试

修改测试代码,打开 ReadOnly 配置(或者 RouteByLatency 和 RouteRandomly 任意一个),观察监控仍然是维持10个 conn,但是按照配置说明,服务是会从读节点读取数据,可以判断 Go-Redis 默认和所有的 Master/Slave 节点都有长连接。

  1. [ec2-user ~]$ netstat -a | grep 6379 | grep ESTABLISHED | awk '{print $5}' | sort | uniq -c | sort -rn
  2.     10 ip-172-31-18-215.a:6379
  3.     10 ip-172-31-46-118.a:6379
  4.     10 ip-172-31-34-217.a:6379
  5.     10 ip-172-31-31-193.a:6379
  6.     10 ip-172-31-15-157.a:6379
  7.     10 ip-172-31-10-163.a:6379

左滑查看更多

参考 CloudWatch Metrics 观察对应的连接数,连接数仍然保持,没有变化,和客户端连接数统计一致。

50208532332ec3132b6ae97d61f10027.png

如果 ReadOnly 打开,在适当压力情况下,观察 GetTypeCmds 监控,可以看到 Master 和 Slave 节点都均匀分布读请求,可以判断读的压力是均匀分配到 Master+Slave 节点上。

0fe3a6eeb3f07c5717526cc6701218ee.png

四、多值查询测试

▌Go-Redis 可以支持在 Redis 非集群和集群模式下 Pipeline 命令正确执行,以下给出 Pipeline 的代码示例:

  1. func TestRedisClusterPipline(t *testing.T) {
  2.    var ctx = context.Background()
  3.    rdb := goredis.NewClusterClient(&goredis.ClusterOptions{
  4. Addrs: []string{"cluster-configuration-endpoint:6379"},
  5. Password: "password",     //密码
  6.        //连接池容量及闲置连接数量
  7.        PoolSize:     10, // 连接池最大socket连接数,默认为4倍CPU数, 4 * runtime.NumCPU
  8.        MinIdleConns: 10, //在启动阶段创建指定数量的Idle连接,并长期维持idle状态的连接数不少于指定数量;。
  9.        //超时
  10.        DialTimeout:  5 * time.Second, //连接建立超时时间,默认5秒。
  11.        ReadTimeout:  3 * time.Second, //读超时,默认3秒, -1表示取消读超时
  12.        WriteTimeout: 3 * time.Second, //写超时,默认等于读超时
  13.        PoolTimeout:  4 * time.Second, //当所有连接都处在繁忙状态时,客户端等待可用连接的最大等待时长,默认为读超时+1秒。
  14.        //闲置连接检查包括IdleTimeout,MaxConnAge
  15.        IdleCheckFrequency: 60 * time.Second, //闲置连接检查的周期,默认为1分钟,-1表示不做周期性检查,只在客户端获取连接时对闲置连接进行处理。
  16.        IdleTimeout:        5 * time.Minute,  //闲置超时,默认5分钟,-1表示取消闲置超时检查
  17.        MaxConnAge:         0 * time.Second,  //连接存活时长,从创建开始计时,超过指定时长则关闭连接,默认为0,即不关闭存活时长较长的连接
  18.        //命令执行失败时的重试策略
  19.        MaxRetries:      10,                     // 命令执行失败时,最多重试多少次,默认为0即不重试
  20.        MinRetryBackoff: 8 * time.Millisecond,   //每次计算重试间隔时间的下限,默认8毫秒,-1表示取消间隔
  21.        MaxRetryBackoff: 512 * time.Millisecond, //每次计算重试间隔时间的上限,默认512毫秒,-1表示取消间隔
  22. TLSConfig: &tls.Config{
  23.          InsecureSkipVerify: true,
  24.        },
  25.        ReadOnly: true,
  26.    })
  27.    rdb.Set(ctx, "test-0", "value-0", 0)
  28.    rdb.Set(ctx, "test-1", "value-1", 0)
  29.    rdb.Set(ctx, "test-2", "value-2", 0)
  30.    pipe := rdb.Pipeline()
  31.    pipe.Get(ctx, "test-0").Result()
  32.    pipe.Get(ctx, "test-1").Result()
  33.    pipe.Get(ctx, "test-2").Result()
  34.    cmders, err := pipe.Exec(ctx)
  35.    if err != nil {
  36.        fmt.Println("err", err)
  37.    }
  38.    for idx, cmder := range cmders {
  39.        cmd := cmder.(*goredis.StringCmd)
  40.        strVal, err := cmd.Result()
  41.        if err != nil {
  42.            fmt.Println("err", err)
  43.        }
  44.        fmt.Println("strVal_" + strconv.Itoa(idx) + ":", strVal)
  45.    }
  46. }

左滑查看更多

▌Go-Redis 不支持 Redis 集群模式下,对不在一个 Shard 的多个 Key 执行 MGet / MSet  操作,如果有类似的使用场景,建议使用 Redis-Go-Cluster 开源项目,源码链接:Redis-Go-Cluster,以下为相应的代码示例。

  1. func TestRedisClusterMGetMSet(t *testing.T) {
  2.    cluster, err := rediscluster.NewCluster(
  3.        &rediscluster.Options{
  4.            StartNodes: []string{"cluster-configuration-endpoint:6379"},
  5.            ConnTimeout: 100 * time.Millisecond,
  6.            ReadTimeout: 100 * time.Millisecond,
  7.            WriteTimeout: 100 * time.Millisecond,
  8.            KeepAlive: 16,
  9.            AliveTime: 60 * time.Second,
  10.        })
  11.    if err != nil {
  12.        fmt.Println(err.Error())
  13.        return
  14.    }
  15.    _, err = cluster.Do("MSET", "test-0", "value-0", "test-1", "value-1", "test-2", "value-2")
  16.    if err != nil {
  17.        fmt.Println("MSET" + err.Error())
  18.        return
  19.    }
  20.    replys, err := rediscluster.Values(cluster.Do("MGET", "test-0", "test-1", "test-2"))
  21.    if err != nil {
  22.        fmt.Println("MGET" + err.Error())
  23.        return
  24.    }
  25.    for i := 0; i < 3; i++ {
  26.        fmt.Println(rediscluster.String(replys[i], nil))
  27.    }
  28. }

左滑查看更多

五、Failover 测试

▌执行 go test 做测试,持续的做读请求

  1. [ec2-user test-redis]$ go test -v redis_test.go -run TestGoRedisCluster -timeout 100m
  2. === RUN   TestGoRedisCluster
  3. 2022-04-18 12:27:37 Job-4 test-0 = value-0-0
  4. 2022-04-18 12:27:37 Job-1 test-0 = value-0-0
  5. 2022-04-18 12:27:37 Job-0 test-0 = value-0-0
  6. 2022-04-18 12:27:37 Job-5 test-0 = value-0-0
  7. 2022-04-18 12:27:37 Job-2 test-0 = value-0-0
  8. 2022-04-18 12:27:37 Job-3 test-0 = value-0-0

左滑查看更多

在 Idle 和 PoolSize 相等的配置下,可以观察到 Redis 客户端服务和 Master 和 Slave 都是建立10个连接。

  1. [ec2-user ~]$ netstat -a | grep 6379 | grep ESTABLISHED | awk '{print $5}' | sort | uniq -c | sort -rn
  2.     10 ip-172-31-18-215.a:6379
  3.     10 ip-172-31-46-118.a:6379
  4.     10 ip-172-31-34-217.a:6379
  5.     10 ip-172-31-31-193.a:6379
  6.     10 ip-172-31-15-157.a:6379
  7.     10 ip-172-31-10-163.a:6379

左滑查看更多

在go test 开始之前,cluster 的 avg 连接数。

76748da901384f1d535376e5ee58a6d8.png

在 go test 执行开始,3个 master 和3个 slave 都新增了10个 conn。

e9950a244f616515281baf03ace04d74.png

▌测试对 redis cluster 的第一个 shard 做主动的 failover

2d6fddaf9c9552808acb32a94b2de4eb.png

▌在命令行输出观察到压测代码发生中断

87f5611bc91c15cc6e315b78564bca0f.png

▌在 ElastiCache Dashboard Events 观察 Failover 过程

41e1e1c9abf09d1b880e156df36eeeb4.png

可以观察到 8:54:13 PM~8:54:36 PM UTC+8 经过 23s 完成 Failover,测试代码的时间戳是12:54:16 ~ 12:54:26 UTC,实际服务中断只有 10s 时间。

在 12:53 UTC 时刻,连接正常。

ca12141811b1e10b9651fdf05b101dd9.png

在 12:54 UTC 时刻,故障节点断开连接。

3a00798e533a8fb189bafc995a277c19.png

在 13:00 UTC 时刻,故障节点开始恢复连接,但是所有服务请求未受到影响。

698d8e8b28e0a9c20ca451f7e1c06983.png

在 13:02 UTC 时刻,所有连接完全恢复。

82c035dc5224132d4c72953973094efc.png

▌在 ReadOnly = False 时,做 Failover 时

  1. 2022-04-18 15:04:27 Job-5 test-1 = value-1-601
  2. 2022-04-18 15:04:27 Job-0 test-1 = value-1-601
  3. 2022-04-18 15:04:27 Job-1 test-1 = value-1-601
  4. err : dial tcp 172.31.31.193:6379: i/o timeouterr : dial tcp 172.31.31.193:6379: i/o timeouterr : dial tcp 172.31.31.193:6379: i/o timeouterr : dial tcp 172.31.31.193:6379: i/o timeouterr : dial tcp 172.31.31.193:6379: i/o timeout2022-04-18 15:04:38 Job-3 test-0 = value-0-603
  5. 2022-04-18 15:04:38 Job-5 test-0 = value-0-603
  6. 2022-04-18 15:04:39 Job-0 test-0 = value-0-603
  7. err : dial tcp 172.31.31.193:6379: i/o timeout2022-04-18 15:04:39 Job-2 test-0 = value-0-603
  8. 2022-04-18 15:04:39 Job-4 test-0 = value-0-603
  9. 2022-04-18 15:04:39 Job-3 test-1 = value-1-604
  10. 2022-04-18 15:04:39 Job-5 test-1 = value-1-604
  11. 2022-04-18 15:04:40 Job-0 test-1 = value-1-604
  12. 2022-04-18 15:04:40 Job-1 test-0 = value-0-603
  13. 2022-04-18 15:04:40 Job-2 test-1 = value-1-604
  14. 2022-04-18 15:04:40 Job-4 test-1 = value-1-604
  15. 2022-04-18 15:04:41 Job-1 test-1 = value-1-604
  16. err : dial tcp 172.31.31.193:6379: i/o timeouterr : dial tcp 172.31.31.193:6379: i/o timeout2022-04-18 15:04:50 Job-3 test-0 = value-0-606
  17. err : dial tcp 172.31.31.193:6379: i/o timeout2022-04-18 15:04:50 Job-2 test-0 = value-0-606
  18. err : dial tcp 172.31.31.193:6379: i/o timeouterr : dial tcp 172.31.31.193:6379: i/o timeouterr : dial tcp 172.31.31.193:6379: i/o timeout2022-04-18 15:04:51 Job-3 test-1 = value-1-607
  19. 2022-04-18 15:04:51 Job-5 test-0 = value-0-606
  20. 2022-04-18 15:04:51 Job-2 test-1 = value-1-607
  21. 2022-04-18 15:04:52 Job-1 test-0 = value-0-606
  22. 2022-04-18 15:04:52 Job-4 test-0 = value-0-606
  23. 2022-04-18 15:04:52 Job-0 test-0 = value-0-606

左滑查看更多

Failover 时,中断时间。

42576fc903c40c13cb8e19a40e005907.png

可以观察到 11:04:28 PM~11:05:03 PM UTC+8 经过 35s 完成 Failover,测试代码的时间戳是 15:04:27 ~ 15:04:51 UTC,实际服务中断为 24s 时间。

a3217e56438476273d9ec5eec3236eb9.png

在 15:03 UTC 连接正常。

70b95d1a89a6c8bfcaa2ecaf0ab0c6a5.png

在 15:04 UTC Failover 开始断开一个节点。

d4304e13f8edee83d0ba6cb94e244b15.png

在 15:11 UTC 开始恢复一个节点。

021ac6a615a84c52ef9cb54553e2088a.png

在 15:13 UTC 完全恢复。

390e9fbda73e4382e843a6118927d5f7.png

六、小结

本博客为大家展示了如何在 Golang 程序中通过 Go-Redis 连接和操作 ElastiCache 集群,从这个简单的 Demo 中我们可以看到 Go-Redis 能很好地支持 ElastiCache 集群开启 TLS 及 Auth 的功能,并自动完成读写分离,负载均衡,Failover 等工作。在第5小节的 Failover 的测试中观察到打开 ReadOnly 可以加速故障恢复,建议实际使用基于 ReadOnly 更好的提升服务读写 Redis Cluster 的性能。通过 Go-Redis,我们可以便捷,高效地使用 ElastiCache 集群。

除了本博客以外,我们还推出了一系列博客,展示了如何在不同语言中使用不同的客户端对 ElastiCache 集群进行连接和操作,欢迎大家阅读。

相关博客

条条大路通罗马 —— 使用 redisson 连接 Amazon ElastiCache for Redis 集群:

https://aws.amazon.com/cn/blogs/china/connecting-amazon-elasticache-for-redis-cluster-using-redisson/

条条大路通罗马 —— 使用 redis-py 访问 Amazon ElastiCache for Redis 集群:

https://aws.amazon.com/cn/blogs/china/use-redis-py-to-access-amazon-elasticache-for-redis-cluster/

条条大路通罗马系列- 使用 Hiredis-cluster 连接 Amazon ElastiCache for Redis 集群:

https://aws.amazon.com/cn/blogs/china/all-roads-to-rome-series-connect-amazon-elasticache-for-redis-cluster-with-hiredis-cluster/

本篇作者

d9b1d95a2d6fc806529bdb2e12046fcb.jpeg

唐健

亚马逊云科技解决方案架构师,负责基于 Amazon 的云计算方案的架构设计,同时致力于 Amazon 云服务在移动应用与互联网行业的应用和推广。拥有多年移动互联网研发及技术团队管理经验,丰富的互联网应用架构项目经历。

907350ad2bea5350dd021b8ab92c6726.gif

d699fac670b0e7ad0f2189659bbe1ff0.gif

听说,点完下面4个按钮

就不会碰到bug了!

e1fab289b8236746313942428ef77fd5.gif

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/繁依Fanyi0/article/detail/511221
推荐阅读
相关标签
  

闽ICP备14008679号