初始化

Author: icowan

.gitignore

@@ -0,0 +1,6 @@
+# Created by .ignore support plugin (hsz.mobi)
+.idea/.gitignore
+.idea/misc.xml
+.idea/modules.xml
+.idea/redis-migrate.iml
+.idea/vcs.xml

Dockerfile

@@ -0,0 +1,14 @@
+/**
+ * @Time: 2020/8/8 16:55
+ * @Author: [email protected]
+ * @File: main
+ * @Software: GoLand
+ */
+
+package main
+
+import "github.com/icowan/redis-migrate/cmd/migrate"
+
+func main() {
+	migrate.Run()
+}

Makefile

@@ -0,0 +1,112 @@
+/**
+ * @Time: 2020/8/8 12:43
+ * @Author: [email protected]
+ * @File: main
+ * @Software: GoLand
+ */
+
+package migrate
+
+import (
+	"flag"
+	"os"
+
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+
+	"github.com/icowan/redis-migrate/redis"
+)
+
+var (
+	sourceHosts, targetHosts, sourceAuth, targetAuth, sourcePrefix, targetPrefix string
+	err                                                                          error
+	sourceRedisCluster, targetRedisCluster                                       bool
+
+	sourceDatabase, targetDatabase int
+
+	sourceRedis redis.RedisInterface
+	targetRedis redis.RedisInterface
+
+	rootCmd = &cobra.Command{
+		Use:               "redis-migrate",
+		Short:             "",
+		SilenceErrors:     true,
+		DisableAutoGenTag: true,
+		Long: `# redis 迁移工具
+可用的配置类型：
+[migrate]
+有关本系统的相关概述，请参阅 http://gitlab.creditease.corp/yqz/invite-admin
+`,
+	}
+
+	migrateRedisCmd = &cobra.Command{
+		Use:               `migrate command <args> [flags]`,
+		Short:             "redis 迁移命令",
+		SilenceErrors:     false,
+		DisableAutoGenTag: false,
+		Example: `
+支持命令:
+[hash, set, sorted-set]
+redis-migrate migrate -h
+`,
+	}
+)
+
+func init() {
+
+	migrateRedisCmd.PersistentFlags().StringVar(&sourceHosts, "source-hosts", "127.0.0.1:6379", "源redis地址, 多个ip用','隔开")
+	migrateRedisCmd.PersistentFlags().StringVar(&targetHosts, "target-hosts", "127.0.0.1:6379", "目标redis地址, 多个ip用','隔开")
+	migrateRedisCmd.PersistentFlags().IntVar(&sourceDatabase, "source-database", 0, "源database")
+	migrateRedisCmd.PersistentFlags().IntVar(&targetDatabase, "target-database", 0, "目标database")
+	migrateRedisCmd.PersistentFlags().StringVar(&sourceAuth, "source-auth", "", "源密码")
+	migrateRedisCmd.PersistentFlags().StringVar(&targetAuth, "target-auth", "", "目标密码")
+	migrateRedisCmd.PersistentFlags().BoolVar(&sourceRedisCluster, "source-redis-cluster", false, "源redis是否是集群")
+	migrateRedisCmd.PersistentFlags().BoolVar(&targetRedisCluster, "target-redis-cluster", false, "目标redis是否是集群")
+	migrateRedisCmd.PersistentFlags().StringVar(&sourcePrefix, "source-prefix", "", "源redis前缀")
+	migrateRedisCmd.PersistentFlags().StringVar(&targetPrefix, "target-prefix", "", "目标redis前缀")
+
+	migrateRedisCmd.AddCommand(migrateRedisHashCmd, migrateRedisSortedSetCmd, migrateRedisSetCmd)
+	addFlags(rootCmd)
+	rootCmd.AddCommand(migrateRedisCmd)
+}
+
+func addFlags(rootCmd *cobra.Command) {
+	flag.CommandLine.VisitAll(func(gf *flag.Flag) {
+		rootCmd.PersistentFlags().AddGoFlag(gf)
+	})
+}
+
+func Run() {
+	if err := rootCmd.Execute(); err != nil {
+		os.Exit(-1)
+	}
+}
+
+func prepare() error {
+	var sourceDrive, targetDrive = redis.RedisSingle, redis.RedisSingle
+	if sourceRedisCluster {
+		sourceDrive = redis.RedisCluster
+	}
+
+	sourceRedis, err = redis.NewRedisClient(sourceDrive, sourceHosts, sourceAuth, sourcePrefix, sourceDatabase)
+	if err != nil {
+		return errors.Wrap(err, "源redis连接失败!")
+	}
+
+	if targetRedisCluster {
+		targetDrive = redis.RedisCluster
+	}
+
+	targetRedis, err = redis.NewRedisClient(targetDrive, targetHosts, targetAuth, targetPrefix, targetDatabase)
+	if err != nil {
+		return errors.Wrap(err, "目标redis连接失败!")
+	}
+	return nil
+}
+
+func getS(n int, char string) (s string) {
+	for i := 1; i <= n; i++ {
+		s += char
+	}
+	return
+}

cmd/main.go

@@ -0,0 +1,88 @@
+/**
+ * @Time: 2020/8/8 16:35
+ * @Author: [email protected]
+ * @File: hash
+ * @Software: GoLand
+ */
+
+package migrate
+
+import (
+	"fmt"
+	"log"
+	"math"
+	"os"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+)
+
+var (
+	migrateRedisHashCmd = &cobra.Command{
+		Use:               `hash <args> [flags]`,
+		Short:             "哈希列表",
+		SilenceErrors:     false,
+		DisableAutoGenTag: false,
+		Example: `
+redis-tool migrate hash {key} --source-redis-cluster true --source-hosts 127.0.0.1:6379,127.0.0.1:7379 --source-auth 123456 --target-redis-cluster true --target-hosts 127.0.0.1:6379,127.0.0.1:7379 --target-auth 123456
+`,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			// 关闭资源连接
+			defer func() {
+				log.Printf("source redis close err: %v", sourceRedis.Close())
+				log.Printf("target redis close err: %v", targetRedis.Close())
+			}()
+			if len(args) < 1 {
+				fmt.Println("至少需要一个参数")
+				return errors.New("参数错误")
+			}
+			return migrateRedisHGetAll(args[0])
+		},
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			if err = prepare(); err != nil {
+				fmt.Println(fmt.Sprintf("prepare error: %s", err.Error()))
+				return err
+			}
+			return nil
+		},
+	}
+)
+
+func migrateRedisHGetAll(key string) error {
+	begin := time.Now()
+
+	total, err := sourceRedis.HLen(key)
+	if err != nil {
+		err = errors.Wrap(err, "sourceRedis.Hlen")
+		return err
+	}
+
+	fmt.Println(fmt.Sprintf("Key: [%s] 总数: [%d]", key, total))
+
+	res, err := sourceRedis.HGetAll(key)
+	if err != nil {
+		err = errors.Wrap(err, "sourceRedis.HGetAll")
+		return err
+	}
+
+	step := math.Ceil(100 / float64(total))
+	var i = 1
+	for k, v := range res {
+		i += int(step)
+		if i >= 100-int(step) {
+			i = 100
+		}
+		_, _ = fmt.Fprintf(os.Stdout, "%d%% [%s]\r", i, getS(i, "#")+getS(100-i, " "))
+		if err = targetRedis.HSet(key, k, v); err != nil {
+			continue
+		}
+		if i == 100 {
+			fmt.Println(fmt.Sprintf("%d%% [%s]\r", i, getS(i, "#")+getS(100-i, " ")))
+		}
+	}
+
+	fmt.Println(fmt.Sprintf("迁移完成, 用时 [%v]", time.Since(begin)))
+
+	return nil
+}

cmd/migrate/migrate.go

@@ -0,0 +1,76 @@
+/**
+ * @Time: 2020/8/8 16:38
+ * @Author: [email protected]
+ * @File: set
+ * @Software: GoLand
+ */
+
+package migrate
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+)
+
+var (
+	migrateRedisSetCmd = &cobra.Command{
+		Use:               `set <args> [flags]`,
+		Short:             "redis set",
+		SilenceErrors:     false,
+		DisableAutoGenTag: false,
+		Example: `
+redis-tool migrate set {key} --source-redis-cluster true --source-hosts 127.0.0.1:6379,127.0.0.1:7379 --source-auth 123456 --target-redis-cluster true --target-hosts 127.0.0.1:6379,127.0.0.1:7379 --target-auth 123456
+`,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			// 关闭资源连接
+			defer func() {
+				log.Printf("source redis close err: %v", sourceRedis.Close())
+				log.Printf("target redis close err: %v", targetRedis.Close())
+			}()
+			if len(args) < 1 {
+				fmt.Println("至少需要一个参数")
+				return errors.New("参数错误")
+			}
+			return migrateRedisSet(args[0])
+		},
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			if err = prepare(); err != nil {
+				fmt.Println(fmt.Sprintf("prepare error: %s", err.Error()))
+				return err
+			}
+			return nil
+		},
+	}
+)
+
+func migrateRedisSet(key string) error {
+	begin := time.Now()
+
+	res, err := sourceRedis.Keys(key)
+	if err != nil {
+		err = errors.Wrap(err, "sourceRedis.Keys")
+		return err
+	}
+
+	fmt.Println(fmt.Sprintf("Key: [%s] 总数: [%d]", key, len(res)))
+
+	for _, val := range res {
+		v, err := sourceRedis.Get(val)
+		if err != nil || v == "" {
+			fmt.Println(fmt.Sprintf("迁移: [%s] --> failure: %s", val, "key不存在"))
+			continue
+		}
+
+		if err = targetRedis.Set(val, v, sourceRedis.TTL(val)); err != nil {
+			fmt.Println(fmt.Sprintf("迁移: [%s] --> failure: %s", val, err.Error()))
+			continue
+		}
+	}
+	fmt.Println(fmt.Sprintf("迁移完成, 用时 [%v]", time.Since(begin)))
+
+	return nil
+}

cmd/migrate/migrate_hash.go

@@ -0,0 +1,92 @@
+/**
+ * @Time: 2020/8/8 16:40
+ * @Author: [email protected]
+ * @File: sorted-set
+ * @Software: GoLand
+ */
+
+package migrate
+
+import (
+	"fmt"
+	"log"
+	"math"
+	"os"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+)
+
+var (
+	migrateRedisSortedSetCmd = &cobra.Command{
+		Use:               `sorted-set <args> [flags]`,
+		Short:             "有序集合",
+		SilenceErrors:     false,
+		DisableAutoGenTag: false,
+		Example: `
+redis-tool migrate sorted-set {key} --source-redis-cluster true --source-hosts 127.0.0.1:6379,127.0.0.1:7379 --source-auth 123456 --target-redis-cluster true --target-hosts 127.0.0.1:6379,127.0.0.1:7379 --target-auth 123456
+`,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			// 关闭资源连接
+			defer func() {
+				log.Printf("source redis close err: %v", sourceRedis.Close())
+				log.Printf("target redis close err: %v", targetRedis.Close())
+			}()
+			if len(args) < 1 {
+				fmt.Println("至少需要一个参数")
+				return errors.New("参数错误")
+			}
+			//return readFile()
+			return migrateRedisSortedSet(args[0])
+		},
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			if err = prepare(); err != nil {
+				fmt.Println(fmt.Sprintf("prepare error: %s", err.Error()))
+				return err
+			}
+			return nil
+		},
+	}
+)
+
+func migrateRedisSortedSet(key string) error {
+	begin := time.Now()
+
+	total, err := sourceRedis.ZCard(key)
+	if err != nil {
+		err = errors.Wrap(err, "sourceRedis.ZCard")
+		return err
+	}
+
+	fmt.Println(fmt.Sprintf("Key: [%s] 总数: [%d]", key, total))
+	var base float64 = 50000
+
+	step := math.Ceil(float64(total) / base)
+
+	for n := 0; n < int(step); n++ {
+		res, err := sourceRedis.ZRangeWithScores(key, int64(n)*50000, int64(n+1)*50000)
+		if err != nil {
+			err = errors.Wrap(err, "sourceRedis.ZRangeWithScores")
+			return err
+		}
+
+		s := 100 / base
+		var i float64 = 0
+		for _, v := range res {
+			i += s
+			if i >= 100-s {
+				i = 100
+			}
+			_, _ = fmt.Fprintf(os.Stdout, "[%d/%d] %.2f%% \r", n+1, int(step), i)
+			if err = targetRedis.ZAdd(key, v.Score, v.Member); err != nil {
+				continue
+			}
+		}
+		fmt.Println(fmt.Sprintf("迁移第: [%d/%d], 用时 [%v]", n+1, int(step), time.Since(begin)))
+	}
+
+	fmt.Println(fmt.Sprintf("迁移完成, 用时 [%v]", time.Since(begin)))
+
+	return nil
+}