注:文中涉及工作环境相关的网址和IP已经被替换

• 服务运行于docker容器中
• 使用kubernetes管理容器
• 服务有多个节点作为一个集群
• 使用rest接口设置服务缓存中的信息
• 需要将信息同步到集群中其他节点

• 使用grpc做服务间通信
• 从etcd中读取服务所有状态为running的节点信息，包括：podIp、status、hostIp、startedAt（启动时间）
• 服务启动时选取运行时间最长的节点，调用grpc接口请求缓存的信息同步到本容器的服务中
• 使用rest接口设置缓存的时候，遍历所有节点（不包括自身），调用grpc接口将信息同步到其他节点

方案特点：

• 不需要借助额外的配置管理工具（如：zookeeper）
• 不需要自行管理节点的配置信息（因为kubernetes的etcd中已经有完整的节点信息）
• grpc开发、传输效率高，扩展性好
• grpc使用http2.0方便后续提供rest接口

#安装grpc依赖库
go get -u google.golang.org/grpc
#安装grpc-go插件，用于将proto文件编译成grpc的golang代码
go get -u github.com/grpc/grpc-go
cd $GOPATH/src

mv github.com/grpc/grpc-go google.golang.org/grpc/grpc-go

遇到go get无法下载的包，也可以通过 http://gopm.io/ 或者 http://golangtc.com/download/package 进行下载

###5. 定义proto文件

syntax = "proto3"; //使用proto3版本

//用于java等语言的package配置
option java_multiple_files = true;
option java_package = "io.grpc.examples.hellorpc";
option java_outer_classname = "hellorpcProto";

//用于golang等语言的package配置
package hellorpc;

//定义服务接口，其中rpc关键字表示 rpc 接口，用于生成grpc接口代码
service Sync {
    rpc Get (SyncRequest) returns(SyncResponse) {}
    rpc Set (SyncRequest) returns(SyncResponse) {}
    rpc GetAll(SyncRequest)returns(SyncResponse) {}
}

//定义请求数据类型, repeated最终会转换成golang中的数组/切片
message SyncRequest {
    repeated SyncData data= 1;
}

//定义返回的数据类型
message SyncResponse {
    repeated SyncData data= 1;
}

//定义实体数据类型，用type字段表示请求的数据类型，用data字段保存请求的数据或者返回的数据
//map<string, string>最终会转换成golang中的map[string]string类型
message SyncData {
    int32 type = 1;
    map<string, string> data = 2;
}

编译proto文件

protoc --go_out=plugins=grpc:./hellorpc hellorpc.proto

其中–go_out用于指定go的proto编译插件以及插件参数
编译成功后，会在 hellorpc目录中生成 hellorpc.pb.go 文件，可以在其他go文件中通过 import “hello-api/hellorpc” 来使用文件中定义的接口

前面提到的 service Sync 部分会编译成如下两部分

type SyncClient interface {
	Get(ctx context.Context, in *SyncRequest, opts ...grpc.CallOption) (*SyncResponse, error)
	Set(ctx context.Context, in *SyncRequest, opts ...grpc.CallOption) (*SyncResponse, error)
	GetAll(ctx context.Context, in *SyncRequest, opts ...grpc.CallOption) (*SyncResponse, error)
}

type SyncServer interface {
	Get(context.Context, *SyncRequest) (*SyncResponse, error)
	Set(context.Context, *SyncRequest) (*SyncResponse, error)
	GetAll(context.Context, *SyncRequest) (*SyncResponse, error)
}

其中 SyncClient 的接口 在 hellorpc.pb.go 里面已经实现好了接口，直接调用即可，但SyncServer定义的接口是需要我们自己实现

//先定义server类型，并实现好SyncServer定义的接口
type server struct {}

const (
    HELLO_SYNC_REST_CLUSTER_INFO = iota
)

func (s *server)Get(ctx context.Context, in *hellorpc.SyncRequest) (*hellorpc.SyncResponse, error){
    var response = hellorpc.SyncResponse{Data: make([]*hellorpc.SyncData, 0, 10)}
    for i := 0; i < len(in.Data); i++{
        request := in.Data[i]
        switch request.Type {
        case hello_SYNC_REST_CLUSTER_INFO:
    		// get something from local cache and set to response
            break
        }
    }
    return &response, nil
}


func (s *server)Set(ctx context.Context, in *hellorpc.SyncRequest) (*hellorpc.SyncResponse, error){
    var response = hellorpc.SyncResponse{Data: make([]*hellorpc.SyncData, 0, 10)}
    for i := 0; i < len(in.Data); i++{
        request := in.Data[i]
        switch request.Type {
        case HELLO_SYNC_REST_CLUSTER_INFO:
    		// set something to local cache, and set the result to response
            break
        }
    }
    return &response, nil
}

func (s *server)GetAll(ctx context.Context, in *hellorpc.SyncRequest) (*hellorpc.SyncResponse, error){
    var response = hellorpc.SyncResponse{Data: make([]*hellorpc.SyncData, 0, 10)}
    for i := 0; i < len(in.Data); i++{
		request := in.Data[i]
        switch request.Type {
        case HELLO_SYNC_REST_CLUSTER_INFO:
    		// get all data from local cache, and set the result to response
            break
        }
    }
    return &response, nil
}

实现好接口后，我们需要将服务注册到grpc，这里我们实现一个名为StartSyncServer的函数来做这些事情

func StartSyncServer(address string) error{
    lis, err := net.Listen("tcp", address)
    if err != nil {
        beego.Debug("start sync server error: %v", err)
        return err
    }
    s := grpc.NewServer()
    hellorpc.RegisterSyncServer(s, &server{})

    //由于s.Serve方法是会一直阻塞住，所以我们需要起一个go routine来执行，在其停止后输出错误信息
    go func(){
        err := s.Serve(lis)
        beego.Debug("sync server stopped with error: %v", err)
    }()
    return nil
}

将StartSyncServer函数添加到模块的 init 函数中执行，我们服务端的代码就基本完成了

//先定义好客户端类型syncClient，这里我们利用继承的方式将hellorpc.SyncClient实现的方法继承过来
type syncClient struct{
	hellorpc.SyncClient
    conn *grpc.ClientConn
    address string
}

func OpenSyncClient(address string)(syncClient, error) {
	s := syncClient{}
    //grpc.WithInsecure用于关闭安全验证，因为我们是在docker内部环境里使用，不暴露在外网，就没有加安全认证了
    conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithTimeout(5*time.Second))
    if err != nil {
        fmt.Println("----open client error %v, conn: %v", err, conn)
        return s, err
    }
    s.conn = conn
    s.address = address
    s.client = hellorpc.NewSyncClient(conn)
    return s, nil
}

func CloseSyncClient(s *syncClient) {
    if s.conn != nil {
        s.conn.Close()
        s.conn = nil
        s.client = nil
    }
}

这样我们只需要编写 c, err := OpenSyncClient(address)，既可通过 response, err := c.Get(context.Background(), request) 的方式调用hellorpc.SyncClient定义的方法

根据etcd的返回值数据结构，我们定义一下两种类型的数据

//用于保存etcd的返回的数据
type EtcdData struct{
    Key             string
    Dir             bool
    Value           interface{}
    CreatedIndex    int32
    ModifiedIndex   int32
    Nodes           []EtcdData
}
//用于保存pod相关的数据
type PodData struct {
    Name            string
    PodIP           string
    HostIP          string
    Status          string
    UpdateTime      string
    Timestamp       int64
}

func newEtcdData() EtcdData{
    return EtcdData{Dir: false, Value: "", Key: "", Nodes: make([]EtcdData, 0, 100)}
}

接下来我们实现EtcdClient

//先定义好EtcdClient的数据结构
type EtcdClient struct{}
//用于解析etcd返回的数据
func parseEtcdData(dataIn map[string]interface{}, dataOut *EtcdData) error {
    if key, ok := dataIn["key"]; ok {
        dataOut.Key = key.(string)
    }
    if isDir, ok := dataIn["dir"]; ok {
        dataOut.Dir = isDir.(bool)
    }
    if value, ok := dataIn["value"]; ok {
        dataOut.Value = value
    }
    if createdIndex, ok := dataIn["createdIndex"]; ok {
        dataOut.CreatedIndex = int32(createdIndex.(float64))
    }
    if modifiedIndex, ok := dataIn["modifiedIndex"]; ok {
        dataOut.ModifiedIndex = int32(modifiedIndex.(float64))
    }
    if nodes, ok := dataIn["nodes"]; ok {
        var subnodes = nodes.([]interface{})
        for i := 0; i < len(subnodes); i++{
            node := subnodes[i].(map[string]interface{})
            var nodeData = newEtcdData()
            parseEtcdData(node, &nodeData)
            dataOut.Nodes = append(dataOut.Nodes,nodeData)
        }
    }
    return nil
}

//实现Get方法用于获取某个key的值
func (c *EtcdClient)Get(baseUrl, key string)(EtcdData, error){
	var url = baseUrl + key
    var res = newEtcdData()
    var result = make(map[string]interface{})
    resp, err := http.Get(url)
    if err == nil{
        out, err1 := ioutil.ReadAll(resp.Body)
        if err1 == nil{
            err2 := json.Unmarshal([]byte(out), &result)
            if err2 != nil{
                return res, err2
            }
            node := result["node"].(map[string]interface{})
            err = parseEtcdData(node, &res)
        }else{
            return res, err1
        }
    }
    return res, err
}

由于我们的服务是跑在docker里，由kubernetes进行服务编排，所以我们需要解析kubernetes在etcd中保存的数据

//用于解析pod的状态信息
func parsePodStatus(podStatus interface{}, podData *PodData){
    pod_status := podStatus.(map[string]interface{}
                            
                            版权声明：本文来源CSDN，感谢博主原创文章，遵循 CC 4.0 by-sa 版权协议，转载请附上原文出处链接和本声明。

                            原文链接：https://blog.csdn.net/letian0805/article/details/72862354

                            站方申明：本站部分内容来自社区用户分享，若涉及侵权，请联系站方删除。