手把手带你使用 go-kit（组件扩充,服务发现）

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Go›Go教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

首先,让我们来回顾一下我们的项目架构

// 项目结构
-| Server
----| server.go
-| EndPoint
----| endpoint.go
-| Transport
----| Transport.go
- main.go

使用外部路由组件扩充服务

我们这里使用 https://github.com/gorilla/mux 很简单这样我们可以使用外部提供的关于路由的功能就可以扩充我们自己的业务逻辑结构(在Transport中修改逻辑)

// main.go
package main

import (
	EndPoint1 "Songzhibin/go-kit-demo/v0/EndPoint"
	"Songzhibin/go-kit-demo/v0/Server"
	"Songzhibin/go-kit-demo/v0/Transport"
	httpTransport "github.com/go-kit/kit/transport/http"
	"github.com/gorilla/mux"
	"net/http"
)

// 服务发布

func main() {
	// 1.先创建我们最开始定义的Server/server.go
	s := Server.Server{}

	// 2.在用EndPoint/endpoint.go 创建业务服务
	hello := EndPoint1.MakeServerEndPointHello(s)
	Bye := EndPoint1.MakeServerEndPointBye(s)

	// 3.使用 kit 创建 handler
	// 固定格式
	// 传入 业务服务 以及 定义的 加密解密方法
	helloServer := httpTransport.NewServer(hello, Transport.HelloDecodeRequest, Transport.HelloEncodeResponse)
	sayServer := httpTransport.NewServer(Bye, Transport.ByeDecodeRequest, Transport.ByeEncodeResponse)

	//// 使用http包启动服务
	//go http.ListenAndServe("0.0.0.0:8000", helloServer)
	//
	//go http.ListenAndServe("0.0.0.0:8001", sayServer)
	//select {}

	// https://github.com/gorilla/mux
	r := mux.NewRouter()
	// 注册路由
	r.Handle("/hello", helloServer)
	r.Handle("/bye", sayServer)
	_ = http.ListenAndServe("0.0.0.0:8000", r)
}

运行一下看下效果

没有问题就可以向外提供关于REST风格的Api了

服务的注册与发现

这里采用的是 `consul` 进行简单Demo 使用其他同理更换api即可

我们使用docker进行部署 https://hub.docker.com/_/consul

// 映射 8500端口 -server 以服务的形式启动 -boostrap 指定自己为leader,而不需要选举 -ui 启动一个内置管理web界面 -client 指定客户端可以访问的ip 0.0.0.0 为任意访问,不设置为127.0.0.1
$ docker pull consul
$ docker run -d --name=Demo -p 8500:8500 consul agent -server -bootstrap -ui -client 0.0.0.0

https://www.consul.io/api-docs/agent/service 查看对应Api

因为我们 `-ui`所以直接在浏览器访问 ip:8500可以看到一个启动的web界面

我们先访问 http://127.0.0.1:8500/v1/agent/services 这时候我们可以看到没有任何服务

这时候我们在程序调用接口进行注册

// main.go
package main

import (
	EndPoint1 "Songzhibin/go-kit-demo/v0/EndPoint"
	"Songzhibin/go-kit-demo/v0/Server"
	"Songzhibin/go-kit-demo/v0/Transport"
	httpTransport "github.com/go-kit/kit/transport/http"
	"github.com/gorilla/mux"
	"net/http"
)

// 服务发布

func main() {
	// 1.先创建我们最开始定义的Server/server.go
	s := Server.Server{}

	// 2.在用EndPoint/endpoint.go 创建业务服务
	hello := EndPoint1.MakeServerEndPointHello(s)
	Bye := EndPoint1.MakeServerEndPointBye(s)

	// 3.使用 kit 创建 handler
	// 固定格式
	// 传入 业务服务 以及 定义的 加密解密方法
	helloServer := httpTransport.NewServer(hello, Transport.HelloDecodeRequest, Transport.HelloEncodeResponse)
	sayServer := httpTransport.NewServer(Bye, Transport.ByeDecodeRequest, Transport.ByeEncodeResponse)

	//// 使用http包启动服务
	//go http.ListenAndServe("0.0.0.0:8000", helloServer)
	//
	//go http.ListenAndServe("0.0.0.0:8001", sayServer)
	//select {}

	// https://github.com/gorilla/mux
	r := mux.NewRouter()
	// 注册路由
	r.Handle("/hello", helloServer)
	r.Handle("/bye", sayServer)

	// 因为这里要做服务发现,所以我们增加一个路由 进行心跳检测使用
	r.Methods("GET").Path("/health").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-type", "application/json")
		w.Write([]byte(`{"status":"ok"}`))
	})
	_ = http.ListenAndServe("0.0.0.0:8000", r)
}

我们新增了一个路由,然后访问试一下效果

然后我们进行服务注册(PS:启动consul服务这台一定要Ping通启动业务服务这台)

我们创建一个xxx.json的文件

{
    "ID":"ServerID", // 唯一
    "Name":"ProjectName", // 不唯一
    "Tags":[
        "primary" // 可设置多Tag
    ],
    "Address": "ip", // 需要注册服务的ip地址
    "Port": 8080, // 注册服务的端口号
    "Check":{
        "HTTP":"url", // 刚才写的健康监测的url
        "interval":"5s" // 探活时间
    }
}

运行指令 这里完全可以使用程序注册哦
$ curl \
      --request PUT \
      --data @xxx.json \
      localhost:8500/v1/agent/service/register

// 取消注册
$ curl \
      --request PUT \
      localhost:8500/v1/agent/service/deregister/注册时候的ID

使用代码进行服务注册发现

https://github.com/hashicorp/consul
https://github.com/hashicorp/consul/tree/master/api // api
我们新建一个 tool目录存放相关配置信息

// 项目结构
-| Server
----| server.go
-| EndPoint
----| endpoint.go
-| Transport
----| Transport.go
-| Tool
----| consul.go
- main.go

// Tool/consul.go
package Tool

import (
	"github.com/hashicorp/consul/api"
)

var client *api.Client
var res api.AgentServiceRegistration

// RegService:服务注册
func RegService(address string, id string, name string, serviceIP string, servicePort int, Interval string, chickHttp string, tag ...string) error {
	// 用于客户端的配置
	config := api.DefaultConfig()
	config.Address = address
	// 用于服务注册
	res = api.AgentServiceRegistration{}
	res.ID = id
	res.Name = name
	res.Address = serviceIP
	res.Port = servicePort
	res.Tags = tag
	chicks := api.AgentServiceCheck{}
	// 间隔时间
	chicks.Interval = Interval
	chicks.HTTP = chickHttp
	// 赋值
	res.Check = &chicks

	// 创建客户端
	var err error
	client, err = api.NewClient(config)
	if err != nil {
		return err
	}
	err = client.Agent().ServiceRegister(&res)
	if err != nil {
		return err
	}
	return nil
}

// LogOutServer 程序关闭后解除注册
func LogOutServer() {
	_ = client.Agent().ServiceDeregister(res.ID)
}

// main.go
package main

import (
	EndPoint1 "Songzhibin/go-kit-demo/v0/EndPoint"
	"Songzhibin/go-kit-demo/v0/Server"
	"Songzhibin/go-kit-demo/v0/Tool"
	"Songzhibin/go-kit-demo/v0/Transport"
	"errors"
	"fmt"
	httpTransport "github.com/go-kit/kit/transport/http"
	"github.com/gorilla/mux"
	"net/http"
	"os"
	"os/signal"
	"syscall"
)

// 服务发布

func main() {
	// 1.先创建我们最开始定义的Server/server.go
	s := Server.Server{}

	// 2.在用EndPoint/endpoint.go 创建业务服务
	hello := EndPoint1.MakeServerEndPointHello(s)
	Bye := EndPoint1.MakeServerEndPointBye(s)

	// 3.使用 kit 创建 handler
	// 固定格式
	// 传入 业务服务 以及 定义的 加密解密方法
	helloServer := httpTransport.NewServer(hello, Transport.HelloDecodeRequest, Transport.HelloEncodeResponse)
	sayServer := httpTransport.NewServer(Bye, Transport.ByeDecodeRequest, Transport.ByeEncodeResponse)

	//// 使用http包启动服务
	//go http.ListenAndServe("0.0.0.0:8000", helloServer)
	//
	//go http.ListenAndServe("0.0.0.0:8001", sayServer)
	//select {}

	// https://github.com/gorilla/mux
	r := mux.NewRouter()
	// 注册路由
	r.Handle("/hello", helloServer)
	r.Handle("/bye", sayServer)
	// 因为这里要做服务发现,所以我们增加一个路由 进行心跳检测使用
	r.Methods("GET").Path("/health").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-type", "application/json")
		_, _ = w.Write([]byte(`{"status":"ok"}`))
	})
	// 注册
	errChan := make(chan error)
	sign := make(chan os.Signal)
	go func() {
		err := Tool.RegService("127.0.0.1:8500", "1", "测试", "127.0.0.1", 8000, "5s", "http://192.168.8.176:8000/health", "test")
		if err != nil {
			errChan <- err
		}
		_ = http.ListenAndServe("0.0.0.0:8000", r)
	}()
	go func() {
		// 接收到信号
		signal.Notify(sign, syscall.SIGINT, syscall.SIGTERM)
		<-sign
		errChan <- errors.New("0")
	}()
	fmt.Println(<-errChan)
	Tool.LogOutServer()
}

客户端使用服务发现调用服务

客户端代码架构看我上一篇客户端直连的博客
主要修改的地方就是Client/client.go 新增了一个方法

// Client/client.go
package Client

import (
	"context"
	"errors"
	"fmt"
	"github.com/go-kit/kit/endpoint"
	"github.com/go-kit/kit/log"
	"github.com/go-kit/kit/sd"
	"github.com/go-kit/kit/sd/consul"
	"github.com/go-kit/kit/transport/http"
	"github.com/hashicorp/consul/api"
	"io"
	"math/rand"
	"net/url"
	"os"
	"strings"
)

// Direct: 直接调用服务端
// method:方法 fullUrl: 完整的url http://localhost:8000
// enc: http.EncodeRequestFunc dec: http.DecodeResponseFunc 这两个函数具体等一下会在Transport中进行详细解释
// requestStruct: 根据EndPoint定义的request结构体传参
func Direct(method string, fullUrl string, enc http.EncodeRequestFunc, dec http.DecodeResponseFunc, requestStruct interface{}) (interface{}, error) {
	// 1.解析url
	target, err := url.Parse(fullUrl)
	if err != nil {
		fmt.Println(err)
		return nil, err
	}
	// kit调用服务端拿到Client对象
	client := http.NewClient(strings.ToUpper(method), target, enc, dec)
	// 调用服务 client.Endpoint()返回一个可执行函数 传入context 和 请求数据结构体
	return client.Endpoint()(context.Background(), requestStruct)
}

// ServiceDiscovery: 通过服务发现的形式调用服务
// registryAddress: 注册中心的地址
// servicesName: 注册的服务名称
// tags: 可用标签
// passingOnly: true 只返回通过健康监测的实例
// method:方法
// enc: http.EncodeRequestFunc dec: http.DecodeResponseFunc 这两个函数具体等一下会在Transport中进行详细解释
// requestStruct: 根据EndPoint定义的request结构体传参
func ServiceDiscovery(method string, registryAddress string, enc http.EncodeRequestFunc, dec http.DecodeResponseFunc, requestStruct interface{}, servicesName string, passingOnly bool, tags ...string) (interface{}, error) {
	// 1.通过consul api创建一个client, 使用go-kit sd 封装一个专门的client 用于获取服务对象
	config := api.DefaultConfig()
	// registryAddress 注册中心的地址
	config.Address = registryAddress

	// 这里是拿到 consul 中的client
	apiClient, err := api.NewClient(config)
	if err != nil {
		return nil, err
	}
	// kit封装client
	client := consul.NewClient(apiClient)

	logger := log.NewLogfmtLogger(os.Stdout)

	// 创建实例
	// logger 可以使用自己的logger对象 例如zap等
	// instances: 实例对象
	instances := consul.NewInstancer(client, logger, servicesName, tags, passingOnly)

	// f: Factory
	// servicesUrl: 传入服务的url, 通过这个url根据直连一样去获取endpoint.Endpoint对象
	f := func(servicesUrl string) (endpoint.Endpoint, io.Closer, error) {
		// 解析url
		target, err := url.Parse("http://" + servicesUrl)
		if err != nil {
			return nil, nil, err
		}
		return http.NewClient(strings.ToUpper(method), target, enc, dec).Endpoint(), nil, nil
	}

	// 获取endpoint可执行对象 与我们直连client.Endpoint()返回的一样
	// 传入 instances: 实例对象  Factory:工厂模式  logger: 日志对象
	endpointer := sd.NewEndpointer(instances, f, logger)

	// 获取所有实例 endpoints
	endpoints, err := endpointer.Endpoints()
	if err != nil {
		return nil, err
	}
	// 随机选择一个执行
	l := len(endpoints)
	if l == 0 {
		return nil, errors.New("len(endpoints) == 0")
	}
	return endpoints[rand.Intn(len(endpoints))](context.Background(), requestStruct)
}

我们调用运行一下

// main.go
package main

import (
	"Songzhibin/go-kit-demo/v0client/Client"
	"Songzhibin/go-kit-demo/v0client/EndPoint"
	"Songzhibin/go-kit-demo/v0client/Transport"
	"fmt"
)

// 调用我们在client封装的函数就好了
func main() {
	//i, err := Client.Direct("GET", "http://127.0.0.1:8000", Transport.ByeEncodeRequestFunc, Transport.ByeDecodeResponseFunc, EndPoint.HelloRequest{Name: "songzhibin"})
	i, err := Client.ServiceDiscovery("GET", "http://127.0.0.1:8500", Transport.ByeEncodeRequestFunc, Transport.ByeDecodeResponseFunc, EndPoint.HelloRequest{Name: "songzhibin"}, "测试", true, "test")
	if err != nil {
		fmt.Println(err)
		return
	}
	res, ok := i.(EndPoint.HelloResponse)
	if !ok {
		fmt.Println("no ok")
		return
	}
	fmt.Println(res)
}

中间件插入

提到中间件大家肯定很熟悉了,在kit中也是一样的采用非侵入式,可定制性非常高,我们来看一下
Ps:中间件的使用在EndPoint中插入
kit中标准中间件函数签名如下

func middleWare(... 自定义参数) endpoint.Middleware

// EndPoint/endpoint.go
package EndPoint

import (
	"Songzhibin/go-kit-demo/v0/Server"
	"context"
	"errors"
	"github.com/go-kit/kit/endpoint"
	"github.com/juju/ratelimit"
	"time"
)

// endpoint.go 定义 Request、Response 格式, 并且可以使用闭包来实现各种中间件的嵌套
// 这里了解 protobuf 的比较好理解点
// 就是声明 接收数据和响应数据的结构体 并通过构造函数创建 在创建的过程当然可以使用闭包来进行一些你想要的操作啦

// 这里根据我们Demo来创建一个响应和请求
// 当然你想怎么创建怎么创建 也可以共用 这里我分开写 便于大家看的清楚

// Hello 业务使用的请求和响应格式
// HelloRequest 请求格式
type HelloRequest struct {
	Name string `json:"name"`
}

// HelloResponse 响应格式
type HelloResponse struct {
	Reply string `json:"reply"`
}

// Bye 业务使用的请求和响应格式
// ByeRequest 请求格式
type ByeRequest struct {
	Name string `json:"name"`
}

// ByeResponse 响应格式
type ByeResponse struct {
	Reply string `json:"reply"`
}

// ------------ 当然 也可以通用的写 ----------
// Request 请求格式
type Request struct {
	Name string `json:"name"`
}

// Response 响应格式
type Response struct {
	Reply string `json:"reply"`
}

type Bucket struct {
	*ratelimit.Bucket
}

// 限流桶
var B Bucket = Bucket{ratelimit.NewBucket(time.Hour, 2)}

// Limiting: 限流
func (b Bucket) Limiting() bool {
	return b.TakeAvailable(1) > 0
}

// MiddleWare: 中间件示例
// 中间件标准函数签名 func middleWare(... 自定义参数) endpoint.Middleware
func MiddleWare(b Bucket) endpoint.Middleware {
	return func(e endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, request interface{}) (response interface{}, err error) {
			// 中间件处理
			if !b.Limiting() {
				return nil, errors.New("false")
			}
			// 如果成功通过就进到下一层
			return e(ctx, request)
		}
	}
}

// 这里创建构造函数 hello方法的业务处理
// MakeServerEndPointHello 创建关于业务的构造函数
// 传入 Server/server.go 定义的相关业务接口
// 返回 go-kit/endpoint.Endpoint (实际上就是一个函数签名)
func MakeServerEndPointHello(s Server.IServer) endpoint.Endpoint {
	// 这里使用闭包,可以在这里做一些中间件业务的处理
	return func(ctx context.Context, request interface{}) (response interface{}, err error) {
		// request 是在对应请求来时传入的参数(这里的request 实际上是等下我们要将的Transport中一个decode函数中处理获得的参数)
		// 这里进行以下断言
		r, ok := request.(HelloRequest)
		if !ok {
			return Response{}, nil
		}
		// 这里实际上就是调用我们在Server/server.go中定义的业务逻辑
		// 我们拿到了 Request.Name 那么我们就可以调用我们的业务 Server.IServer 中的方法来处理这个数据并返回
		// 具体的业务逻辑具体定义....
		return HelloResponse{Reply: s.Hello(r.Name)}, nil
		// response 这里返回的response 可以返回任意的 不过根据规范是要返回我们刚才定义好的返回对象

	}
}

// 这里创建构造函数 Bye方法的业务处理
// MakeServerEndPointBye 创建关于业务的构造函数
// 传入 Server/server.go 定义的相关业务接口
// 返回 go-kit/endpoint.Endpoint (实际上就是一个函数签名)
func MakeServerEndPointBye(s Server.IServer) endpoint.Endpoint {
	// 这里使用闭包,可以在这里做一些中间件业务的处理
	return func(ctx context.Context, request interface{}) (response interface{}, err error) {
		// request 是在对应请求来时传入的参数(这里的request 实际上是等下我们要将的Transport中一个decode函数中处理获得的参数)
		// 这里进行以下断言
		r, ok := request.(ByeRequest)
		if !ok {
			return Response{}, nil
		}
		// 这里实际上就是调用我们在Server/server.go中定义的业务逻辑
		// 我们拿到了 Request.Name 那么我们就可以调用我们的业务 Server.IServer 中的方法来处理这个数据并返回
		// 具体的业务逻辑具体定义....
		return ByeResponse{Reply: s.Bye(r.Name)}, nil
		// response 这里返回的response 可以返回任意的 不过根据规范是要返回我们刚才定义好的返回对象
	}
}

在mian.go中也做了一下嵌套

// main.go
package main

import (
	EndPoint1 "Songzhibin/go-kit-demo/v0/EndPoint"
	"Songzhibin/go-kit-demo/v0/Server"
	"Songzhibin/go-kit-demo/v0/Tool"
	"Songzhibin/go-kit-demo/v0/Transport"
	"errors"
	"fmt"
	httpTransport "github.com/go-kit/kit/transport/http"
	"github.com/gorilla/mux"
	"net/http"
	"os"
	"os/signal"
	"syscall"
)

// 服务发布

func main() {
	// 1.先创建我们最开始定义的Server/server.go
	s := Server.Server{}

	// 2.在用EndPoint/endpoint.go 创建业务服务

	hello := EndPoint1.MakeServerEndPointHello(s)

	// 加入中间件
	Bye := EndPoint1.MiddleWare(EndPoint1.B)(EndPoint1.MakeServerEndPointBye(s))

	// 3.使用 kit 创建 handler
	// 固定格式
	// 传入 业务服务 以及 定义的 加密解密方法
	helloServer := httpTransport.NewServer(hello, Transport.HelloDecodeRequest, Transport.HelloEncodeResponse)
	sayServer := httpTransport.NewServer(Bye, Transport.ByeDecodeRequest, Transport.ByeEncodeResponse)

	//// 使用http包启动服务
	//go http.ListenAndServe("0.0.0.0:8000", helloServer)
	//
	//go http.ListenAndServe("0.0.0.0:8001", sayServer)
	//select {}

	// https://github.com/gorilla/mux
	r := mux.NewRouter()
	// 注册路由
	r.Handle("/hello", helloServer)
	r.Handle("/bye", sayServer)
	// 因为这里要做服务发现,所以我们增加一个路由 进行心跳检测使用
	r.Methods("GET").Path("/health").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-type", "application/json")
		_, _ = w.Write([]byte(`{"status":"ok"}`))
	})
	// 注册
	errChan := make(chan error)
	sign := make(chan os.Signal)
	go func() {
		err := Tool.RegService("127.0.0.1:8500", "1", "测试", "127.0.0.1", 8000, "5s", "http://10.43.1.106:8000/health", "test")
		if err != nil {
			errChan <- err
		}
		_ = http.ListenAndServe("0.0.0.0:8000", r)
	}()
	go func() {
		// 接收到信号
		signal.Notify(sign, syscall.SIGINT, syscall.SIGTERM)
		<-sign
		errChan <- errors.New("0")
	}()
	fmt.Println(<-errChan)
	Tool.LogOutServer()
}