一、目標

二、使用步驟

1. 安裝 consul

我們可以直接使用官方提供的二進制文件來進行安裝部署，其官網地址為 https://www.consul.io/downloads

下載后為可執行文件，在我們開發試驗過程中，可以直接使用 consul agent -dev 命令來啟動一個單節點的 consul

在啟動的打印日志中可以看到 agent: Started HTTP server on 127.0.0.1:8500 (tcp), 我們可以在瀏覽器直接訪問 127.0.0.1:8500 即可看到如下

這里我們的 consul 就啟動成功了

2. 服務注冊

在網絡編程中，一般會提供項目的 IP、PORT、PROTOCOL，在服務治理中，我們還需要知道對應的服務名、實例名以及一些自定義的擴展信息

在這里使用 ServiceInstance 接口來規定注冊服務時必須的一些信息，同時用 DefaultServiceInstance 實現

type ServiceInstance interface { 
    // return The unique instance ID as registered. 
    GetInstanceId() string 
    // return The service ID as registered. 
    GetServiceId() string 
    // return The hostname of the registered service instance. 
    GetHost() string 
    // return The port of the registered service instance. 
    GetPort() int    // return Whether the port of the registered service instance uses HTTPS. 
    IsSecure() bool    // return The key / value pair metadata associated with the service instance. 
    GetMetadata() map[string]string 
}type DefaultServiceInstance struct { 
    InstanceId string 
    ServiceId  string 
    Host       string 
    Port       int    Secure     bool    Metadata   map[string]string 
}func NewDefaultServiceInstance(serviceId string, host string, port int, secure bool, 
    metadata map[string]string, instanceId string) (*DefaultServiceInstance, error) { 
    // 如果沒有傳入 IP 則獲取一下，這個方法在多網卡的情況下，并不好用    if len(host) == 0 { 
        localIP, err := util.GetLocalIP()        if err != nil { 
            return nil, err 
        }        host = localIP    }    if len(instanceId) == 0 { 
        instanceId = serviceId + "-" + strconv.FormatInt(time.Now().Unix(), 10) + "-" + strconv.Itoa(rand.Intn(9000)+1000) 
    }    return &DefaultServiceInstance{InstanceId: instanceId, ServiceId: serviceId, Host: host, Port: port, Secure: secure, Metadata: metadata}, nil 
}func (serviceInstance DefaultServiceInstance) GetInstanceId() string { 
    return serviceInstance.InstanceId 
}func (serviceInstance DefaultServiceInstance) GetServiceId() string { 
    return serviceInstance.ServiceId 
}func (serviceInstance DefaultServiceInstance) GetHost() string { 
    return serviceInstance.Host 
}func (serviceInstance DefaultServiceInstance) GetPort() int {    return serviceInstance.Port 
}func (serviceInstance DefaultServiceInstance) IsSecure() bool {    return serviceInstance.Secure 
}func (serviceInstance DefaultServiceInstance) GetMetadata() map[string]string { 
    return serviceInstance.Metadata 
} 

定義接口

在上面規定了需要注冊的服務的必要信息，下面定義下服務注冊和剔除的方法

type ServiceRegistry interface { 
    Register(serviceInstance cloud.ServiceInstance) bool 
    Deregister() 
} 

具體實現

因為 consul 提供了 http 接口來對 consul 進行操作，我們也可以使用 http 請求方式進行注冊和剔除操作，具體 http 接口文檔見 https://www.consul.io/api-docs, consul 默認提供了go 語言的實現，這里直接使用 github.com/hashicorp/consul/api

import ( 
    "errors" 
    "fmt" 
    "github.com/hashicorp/consul/api" 
    "strconv" 
    "unsafe" 
)type consulServiceRegistry struct { 
    serviceInstances     map[string]map[string]cloud.ServiceInstance 
    client               api.Client    localServiceInstance cloud.ServiceInstance}func (c consulServiceRegistry) Register(serviceInstance cloud.ServiceInstance) bool {    // 創建注冊到consul的服務到    registration := new(api.AgentServiceRegistration)    registration.ID = serviceInstance.GetInstanceId()    registration.Name = serviceInstance.GetServiceId()    registration.Port = serviceInstance.GetPort()    var tags []string 
    if serviceInstance.IsSecure() { 
        tags = append(tags, "secure=true") 
    } else { 
        tags = append(tags, "secure=false") 
    }    if serviceInstance.GetMetadata() != nil { 
        var tags []string 
        for key, value := range serviceInstance.GetMetadata() { 
            tags = append(tags, key+"="+value) 
        }        registration.Tags = tags    }    registration.Tags = tags    registration.Address = serviceInstance.GetHost()    // 增加consul健康檢查回調函數    check := new(api.AgentServiceCheck)    schema := "http" 
    if serviceInstance.IsSecure() { 
        schema = "https" 
    }    check.HTTP = fmt.Sprintf("%s://%s:%d/actuator/health", schema, registration.Address, registration.Port) 
    check.Timeout = "5s" 
    check.Interval = "5s" 
    check.DeregisterCriticalServiceAfter = "20s" // 故障檢查失敗30s后 consul自動將注冊服務刪除 
    registration.Check = check    // 注冊服務到consul    err := c.client.Agent().ServiceRegister(registration)    if err != nil { 
        fmt.Println(err)        return false 
    }    if c.serviceInstances == nil { 
        c.serviceInstances = map[string]map[string]cloud.ServiceInstance{} 
    }    services := c.serviceInstances[serviceInstance.GetServiceId()]    if services == nil { 
        services = map[string]cloud.ServiceInstance{} 
    }    services[serviceInstance.GetInstanceId()] = serviceInstance    c.serviceInstances[serviceInstance.GetServiceId()] = services    c.localServiceInstance = serviceInstance    return true 
}// deregister a servicefunc (c consulServiceRegistry) Deregister() {    if c.serviceInstances == nil { 
        return 
    }    services := c.serviceInstances[c.localServiceInstance.GetServiceId()]    if services == nil { 
        return 
    }    delete(services, c.localServiceInstance.GetInstanceId())    if len(services) == 0 { 
        delete(c.serviceInstances, c.localServiceInstance.GetServiceId())    }    _ = c.client.Agent().ServiceDeregister(c.localServiceInstance.GetInstanceId())    c.localServiceInstance = nil 
}// new a consulServiceRegistry instance// token is optionalfunc NewConsulServiceRegistry(host string, port int, token string) (*consulServiceRegistry, error) { 
    if len(host) < 3 { 
        return nil, errors.New("check host") 
    }    if port <= 0 || port > 65535 { 
        return nil, errors.New("check port, port should between 1 and 65535") 
    }    config := api.DefaultConfig() 
    config.Address = host + ":" + strconv.Itoa(port) 
    config.Token = token 
    client, err := api.NewClient(config) 
    if err != nil { 
        return nil, err 
    }    return &consulServiceRegistry{client: *client}, nil 
} 

測試用例

注冊服務的代碼基本完成，來測試一下

func TestConsulServiceRegistry(t *testing.T) { 
    host := "127.0.0.1" 
    port := 8500 
    registryDiscoveryClient, _ := extension.NewConsulServiceRegistry(host, port, "") 
    ip, err := util.GetLocalIP()    if err != nil { 
        t.Error(err)    }    serviceInstanceInfo, _ := cloud.NewDefaultServiceInstance("go-user-server", "", 8090, 
        false, map[string]string{"user":"zyn"}, "") 
    registryDiscoveryClient.Register(serviceInstanceInfo)    r := gin.Default()    // 健康檢測接口，其實只要是 200 就認為成功了 
    r.GET("/actuator/health", func(c *gin.Context) { 
        c.JSON(200, gin.H{ 
            "message": "pong", 
        }) 
    }) 
    err = r.Run(":8090") 
    if err != nil{ 
        registryDiscoveryClient.Deregister() 
    } 
} 

如果成功，則會在 consul 看到 go-user-server 這個服務

3. 服務發現

在服務發現中，一般會需要兩個方法

• 獲取所有的服務列表
• 獲取指定的服務的所有實例信息

接口定義

type DiscoveryClient interface { 
    /** 
     * Gets all ServiceInstances associated with a particular serviceId. 
     * @param serviceId The serviceId to query. 
     * @return A List of ServiceInstance. 
     */ 
    GetInstances(serviceId string) ([]cloud.ServiceInstance, error)    /** 
     * @return All known service IDs. 
     */ 
    GetServices() ([]string, error)} 

具體實現

來實現一下

type consulServiceRegistry struct { 
    serviceInstances     map[string]map[string]cloud.ServiceInstance 
    client               api.Client    localServiceInstance cloud.ServiceInstance}func (c consulServiceRegistry) GetInstances(serviceId string) ([]cloud.ServiceInstance, error) { 
    catalogService, _, _ := c.client.Catalog().Service(serviceId, "", nil) 
    if len(catalogService) > 0 { 
        result := make([]cloud.ServiceInstance, len(catalogService)) 
        for index, sever := range catalogService { 
            s := cloud.DefaultServiceInstance{                InstanceId: sever.ServiceID,                ServiceId:  sever.ServiceName,                Host:       sever.Address,                Port:       sever.ServicePort,                Metadata:   sever.ServiceMeta,            }            result[index] = s        }        return result, nil 
    }    return nil, nil 
}func (c consulServiceRegistry) GetServices() ([]string, error) { 
    services, _, _ := c.client.Catalog().Services(nil) 
    result := make([]string, unsafe.Sizeof(services)) 
    index := 0 
    for serviceName, _ := range services { 
        result[index] = serviceName        index++    }    return result, nil 
}// new a consulServiceRegistry instance 
// token is optional 
func NewConsulServiceRegistry(host string, port int, token string) (*consulServiceRegistry, error) { 
    if len(host) < 3 { 
        return nil, errors.New("check host") 
    } 
    if port <= 0 || port > 65535 { 
        return nil, errors.New("check port, port should between 1 and 65535") 
    } 
    config := api.DefaultConfig() 
    config.Address = host + ":" + strconv.Itoa(port) 
    config.Token = token 
    client, err := api.NewClient(config) 
    if err != nil { 
        return nil, err 
    } 
    return &consulServiceRegistry{client: *client}, nil 
} 

測試用例

func TestConsulServiceDiscovery(t *testing.T) { 
    host := "127.0.0.1" 
    port := 8500 
    token := "" 
    registryDiscoveryClient, err := extension.NewConsulServiceRegistry(host, port, token) 
    if err != nil {        panic(err) 
    }    t.Log(registryDiscoveryClient.GetServices()) 
    t.Log(registryDiscoveryClient.GetInstances("go-user-server")) 
} 

結果

consul_service_registry_test.go:57: [consul go-user-server      ] <nil> 
consul_service_registry_test.go:59: [{go-user-server-1602590661-56179 go-user-server 127.0.0.1 8090 false map[user:zyn]}] <nil> 

總結

通過使用 consul api 我們可以簡單的實現基于 consul 的服務發現，在通過結合 http rpc 就可簡單的實現服務的調用，下面一章來簡單講下 go 如何發起 http 請求，為我們做 rpc 做個鋪墊