实战分享：利用go-zero实现微服务的数据迁移

syntax = "proto3";

package user;

option go_package = "github.com/xiaoyu-eric/go-zero/userinfo/proto/user";

message User {
  string id = 1;
  string name = 2;
  string mobile = 3;
  string email = 4;
  string address = 5;
  int32 age = 6;
}

syntax = "proto3";

package user;

option go_package = "github.com/xiaoyu-eric/go-zero/userinfo/proto/user";

service UserSvc {
  rpc List(UserListReq) returns (UserListReply);
}

message User {
  string id = 1;
  string name = 2;
  string mobile = 3;
  string email = 4;
  string address = 5;
  int32 age = 6;
}

message UserListReq {}

message UserListReply {
  repeated User list = 1;
}

package main

import (
    "context"
    "database/sql"
    "fmt"
    "log"

    "github.com/tal-tech/go-zero/core/stores/sqlx"
    "github.com/tal-tech/go-zero/core/syncx"
    "github.com/tal-tech/go-zero/core/threading"
    "github.com/xiaoyu-eric/go-zero/userinfo/proto/user"
    "github.com/xiaoyu-eric/go-zero/userinfo/types"
    "google.golang.org/grpc"
)

const (
    sourceAddress     = "localhost:8888"
    targetAddress     = "localhost:3306"
    targetUser        = "root"
    targetPassword    = "123456"
    targetDatabase    = "userinfo"
    concurrency       = 10
    targetTablePrefix = "user_"
)

func main() {
    var (
        dbTarget     = initDb(targetAddress, targetUser, targetPassword, targetDatabase)
        usersChannel = make(chan []*user.User, concurrency)
        wg           = syncx.WaitGroup{}
    )

    defer func() {
        if err := dbTarget.Close(); err != nil {
            log.Fatalf("Failed to close dbTarget: %v", err)
        }
    }()

    conn, err := grpc.Dial(sourceAddress, grpc.WithInsecure())
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    defer conn.Close()

    client := user.NewUserSvcClient(conn)

    req := &user.UserListReq{}
    stream, err := client.List(context.Background(), req)
    if err != nil {
        log.Fatalf("Failed to call List: %v", err)
    }

    wg.Add(concurrency)
    go func() {
        for {
            users, err := stream.Recv()
            if err != nil {
                close(usersChannel)
                break
            }
            if len(users.List) > 0 {
                usersChannel <- users.List
            }
        }
        wg.Done()
    }()

    for i := 0; i < concurrency; i++ {
        threading.GoSafe(func() {
            for {
                users, ok := <-usersChannel
                if !ok {
                    break
                }
                if err := insertUsers(types.Users(users), dbTarget); err != nil {
                    log.Fatalf("Failed to insert users: %v", err)
                }
            }
            wg.Done()
        })
    }

    wg.Wait()
}

func insertUsers(users types.Users, db *sqlx.DB) error {
    tx, err := db.Beginx()
    if err != nil {
        return fmt.Errorf("Failed to begin transaction: %v", err)
    }

    for _, user := range users {
        if _, err := tx.NamedExec(`INSERT INTO `+targetTablePrefix+`(id, name, mobile, email, address, age) VALUES(:id, :name, :mobile, :email, :address, :age)`, user); err != nil {
            if rbErr := tx.Rollback(); rbErr != nil {
                return fmt.Errorf("Failed to Rollback: %v", rbErr)
            }
            return fmt.Errorf("Failed to insert user: %v", err)
        }
    }

    if err := tx.Commit(); err != nil {
        return fmt.Errorf("Failed to commit transaction: %v", err)
    }

    return nil
}

func initDb(address, username, password, database string) *sqlx.DB {
    dataSourceName := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4&parseTime=True&loc=Local", username, password, address, database)
    conn := sqlx.NewMysql(dataSourceName)
    return conn
}

#!/bin/bash

if [ $# -lt 1 ]; then
    echo "Usage: $0 [microservice]"
    exit 1
fi

microservice=$1
path="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"

cd $path/$microservice/cmd/transfer && go build -o transfer && 
./transfer