Golang微服务开发实践

github: github.com/yun-mu/Micr…

微服务概念学习：可参考 Nginx 的微服务文章

微服务最佳实践：可参考微服务最佳实践

服务：

<ul><li>consignment-service（货运服务）</li><li>user-service（用户服务）</li><li>log-service (日志服务)</li><li>vessel-service（货船服务）</li><li>api-service (API 服务)</li></ul>

用到的技术栈如下：

<pre><code class="hljs yaml copyable">framework: go-micro, gin Transport: tcp Server: rpc Client: rpc RegisterTTL: 30s RegisterInterval: 20s Registry: consul, 服务发现和注册 Broker: kafka, 消息队列 Selector: cache, 负载均衡 Codec: protobuf, 编码 Tracing: jaeger, 链路追踪 Metrics: jaeger breaker: hystrix, 熔断 ratelimit: uber/ratelimit, 限流 复制代码</code></pre> <h3 class="heading">服务关系图</h3>

<h3 class="heading">实体关系图</h3>

<h3 class="heading">服务流程示例</h3>

采用 JWT

<h3 class="heading">发布订阅模式</h3>

前提工具：<code>go, dep, docker, docker-compose, mongo</code>

首先初始化：<code>make init</code>

Makefile 部分代码如下：

<pre><code class="hljs makefile copyable">init: cd .. mv MicroServicePractice ${GOPATH}/src/Ethan/ ./pull.sh # 安装 go 依赖 cd plugins docker-compose -f docker-compose.yml up -d # 安装插件，如：kafka, consul, zookeeper, jaeger 复制代码</code></pre>

之后就可以运行代码了：

注：建议自己开多个终端 <code>go run</code> ，这样可以看日志

<pre><code class="hljs shell copyable">make run # 允许服务 server 复制代码</code></pre>

测试：

注：注意顺序，刚开始啥数据都没有的

<pre><code class="hljs shell copyable">go run user-cli/cli.go export Token=$Token # 注意换成前面生成的Token go run vessel-cli/cli.go go run consignment-cli/cli.go 复制代码</code></pre>

<h2 class="heading">开发详解</h2> <h3 class="heading">proto 代码生成</h3>

安装工具：

protoc 安装：google.github.io/proto-lens/…

protoc-gen-go 和 protoc-gen-micro

<pre><code class="hljs shell copyable">go get -u -v google.golang.org/grpc go get -u -v github.com/golang/protobuf/protoc-gen-go go get -u -v github.com/micro/protoc-gen-micro 复制代码</code></pre>

生成的脚本我已经写好 Makefile, 进入 interface-center 目录，执行<code>make build</code> 即可

内部示例如下：

<pre><code class="hljs shell copyable">protoc --proto_path=proto:. --go_out=plugins=micro:out/ proto/vessel/vessel.proto 复制代码</code></pre>

这里使用 micro 插件，若想和不使用插件对比，可使用如下命令：

<pre><code class="hljs shell copyable">protoc --proto_path=proto:. --go_out=out/ --micro_out=out/ proto/vessel/vessel.proto 复制代码</code></pre>

这样会生成两个文件，一个为 .micro.go 一个为 .pb.go

这里顺便看一下生成的 pb 文件里是如何进行 rpc 调用的，我们随便看一个方法，如：vessel 的 FindAvailable

<pre><code class="hljs go copyable" lang="go">func (c *vesselServiceClient) FindAvailable(ctx context.Context, in *Specification, opts ...client.CallOption) (*Response, error) { req := c.c.NewRequest(c.serviceName, "VesselService.FindAvailable", in) out := new(Response) err := c.c.Call(ctx, req, out, opts...) if err != nil { return nil, err } return out, nil } 复制代码</code></pre> <pre><code class="hljs bash copyable">func newRequest(service, endpoint string, request interface{}, contentType string, reqOpts ...RequestOption) Request { var opts RequestOptions for _, o := range reqOpts { o(&opts) } // set the content-type specified if len(opts.ContentType) > 0 { contentType = opts.ContentType } return &rpcRequest{ service: service, method: endpoint, endpoint: endpoint, body: request, contentType: contentType, opts: opts, } } 复制代码</code></pre> <h3 class="heading">微服务开发流程</h3>

如果使用 grpc 作为 server 和 client，开发流程如下：

注：server 和 client 必须相同，如：我的代码中 server 和 client 使用的都是 rpc, transport 是 tcp

<h3 class="heading">目录简介</h3> <ul><li>api：对外暴露的HTTP web 接口，可以理解为网关</li><li>common：所有服务都能调用的东西，如 GetMicroClient, GetMicroServer</li><li>config：配置中心，其他服务的启动都依赖的配置</li><li>consignment</li><li>consignment-cli：cli 测试</li><li>interface-center：proto 文件中心，同时生成的 .go 文件也在这里</li><li>shippy-ui：前端测试 ui 代码，对接API，API还没写完</li><li>user</li><li>user-cli</li><li>vessel</li><li>vessel-cli</li></ul><h3 class="heading">初始化</h3>

示例代码：consignment/main.go, common/service.go

<pre><code class="hljs go copyable" lang="go">// 直接调用自己写的公有的库获取 server，保持配置同步 // common.AuthWrapper 为前置认证，采用JWT srv := common.GetMicroServer(service, micro.WrapHandler(common.AuthWrapper)) 复制代码</code></pre>

common.GetMicroServer

<pre><code class="hljs go copyable" lang="go">func GetMicroServer(service string, exOpts ...micro.Option) micro.Service { opts := getOpts(service) if defaultServer != nil { opts = append(opts, defaultServer) } // ... // 注意顺序，同样的配置后面的会将前面的覆盖 opts = append(opts, exOpts...) srv := micro.NewService(opts...) // 初始化，解析命令行参数 srv.Init() return srv } 复制代码</code></pre>

注：调用者的 client, transport 应当和 server 的 client, transport 配置相同，所以开发 micro web的时候要注意！micro web 全是 HTTP 或者 ws，需要自己使用和后面服务相同的 client 来完成转发。

这里我的demo中采用了 consul，consul 自带了 UI和健康检查，consul UI 端口为：8500

<pre><code class="hljs go copyable" lang="go">// 注册延迟，30s 内没有注册则失效，consul 会自动删除服务 micro.RegisterTTL(time.Second * 30), // 注册间隔，每隔 20s 注册一次 micro.RegisterInterval(time.Second * 20) // ... // opts 中添加如下配置即可 micro.Registry(consul.NewRegistry(func(op *registry.Options) { op.Addrs = config.GetRegistryAddrs(service) })) 复制代码</code></pre>

docker-compose.yml 中已经定义，这里测试用，因此只采用单节点(server)的形式，consul 采用 Raft 算法，为了保证选主无误，节点(server)数必须是奇数，bootstrap-expect 表示节点数量

<pre><code class="hljs yaml copyable"> consul: image: consul:1.5 container_name: consul-node1 command: agent -server -bootstrap-expect=1 -node=node1 -bind=0.0.0.0 -client=0.0.0.0 -datacenter=dc1 volumes: - ./consul/node1:/consul/data consul-client: image: consul:1.5 container_name: consul-client1 command: agent -retry-join=consul -node=client1 -bind=0.0.0.0 -client=0.0.0.0 -datacenter=dc1 -ui ports: - "8500:8500" - "8600:8600" - "8300:8300" depends_on: - consul volumes: - ./consul/client1:/consul/data 复制代码</code></pre> <h3 class="heading">watch</h3> <pre><code class="hljs go copyable" lang="go"> // 这里的 handler 应当实现 pb 中定义的调用 h := handler.GetHandler(session, vClient, uClient, bk) // 将 server 作为微服务的服务端 pb.RegisterShippingServiceHandler(srv.Server(), h) if err := srv.Run(); err != nil { log.Fatalf("failed to serve: %v", err) } 复制代码</code></pre>

<pre><code class="hljs protobuf copyable">// 货轮微服务 service ShippingService { // 托运一批货物 rpc CreateConsignment (Consignment) returns (Response) { } // 查看托运货物的信息 rpc GetConsignments (GetRequest) returns (Response) { } } 复制代码</code></pre>

handler

<pre><code class="hljs go copyable" lang="go">func (h *Handler) CreateConsignment(ctx context.Context, req *pb.Consignment, resp *pb.Response) error { } func (h *Handler) GetConsignments(ctx context.Context, req *pb.GetRequest, resp *pb.Response) error { } 复制代码</code></pre> <h3 class="heading">上下文 context</h3>

在我们的 AuthWrapper 中，ctx 作为上下文信息传递的方式，可在 ctx 中添加信息

<pre><code class="hljs go copyable" lang="go">// // AuthWrapper 是一个高阶函数，入参是 "下一步" 函数，出参是认证函数 // 在返回的函数内部处理完认证逻辑后，再手动调用 fn() 进行下一步处理 // token 是从上下文中取出的，再调用 user-service 将其做验证 // 认证通过则 fn() 继续执行，否则报错 // func AuthWrapper(fn server.HandlerFunc) server.HandlerFunc { log.Println("AuthWrapper") return func(ctx context.Context, req server.Request, resp interface{}) error { // consignment-service 独立测试时不进行认证 if os.Getenv("DISABLE_AUTH") == "true" { return fn(ctx, req, resp) } meta, ok := metadata.FromContext(ctx) if !ok { return errors.New("no auth meta-data found in request") } token := meta["token"] // Auth here authResp, err := GetUserClient().ValidateToken(context.Background(), &userPb.Token{ Token: token, }) log.Println("Auth Resp:", authResp) if err != nil { return err } // 这里将 JWT 解析出来的 user_id 传递下去 ctx = context.WithValue(ctx, "user_id", authResp.UserId) err = fn(ctx, req, resp) return err } } 复制代码</code></pre> <h3 class="heading">handler</h3>

在所有前置操作执行完毕之后，开始执行 handler 真正的 Call，handler 的函数定义必须和 pb 中一模一样。

处理完之后直接编辑 resp 即可，之后返回 nil，resp 是一个指针，直接传递了返回信息。

<pre><code class="hljs go copyable" lang="go">func (h *Handler) CreateConsignment(ctx context.Context, req *pb.Consignment, resp *pb.Response) error { // ... 处理 resp.Created = true resp.Consignment = req // 后置操作 go func() { // ... h.pubLog(userID, "CreateConsignment", msg) }() return nil } 复制代码</code></pre> <h3 class="heading">过滤器</h3>

这里以版本过滤器为例：

<pre><code class="hljs go copyable" lang="go">// Filter will filter the version of the service func Filter(v string) client.CallOption { if v == "" { v = "latest" } filter := func(services []*registry.Service) []*registry.Service { var filtered []*registry.Service for _, service := range services { if service.Version == v { filtered = append(filtered, service) } } return filtered } return client.WithSelectOption(selector.WithFilter(filter)) } 复制代码</code></pre>

之后在进行 client.Call 的时候可以使用

<pre><code class="hljs go copyable" lang="go">vResp, err := h.vesselClient.FindAvailable(ctx, vReq, common.Filter(version)) 复制代码</code></pre> <h3 class="heading">db 交互</h3>

一般而言不要把 pb 结构体直接插入数据库中，最好有一个中间转换层。示例如下：

<pre><code class="hljs go copyable" lang="go">func (repo *ConsignmentRepository) Create(con *pb.Consignment) error { // 这里将PB转换为想要的结构体，之后再插入 data := PBConsignment2Consignment(con) // dao 层直接对接DB操作 return dao.Insert(repo.collection(), &data) } // 在外面记得把 Session Close 复制代码</code></pre> <h3 class="heading">broker</h3>

消息队列

<pre><code class="hljs go copyable" lang="go">// common/service.go/GetMicroServer() // 注册 brokerKafka := kafka.NewBroker(func(options *broker.Options) { // eg: []{"127.0.0.1:9092"} options.Addrs = config.GetBrokerAddrs(service) }) if err := brokerKafka.Connect(); err != nil { log.Fatalf("Broker Connect error: %v", err) } // ... micro.Broker(brokerKafka) 复制代码</code></pre>

注册完之后就开始定义接口了

<pre><code class="hljs go copyable" lang="go"> // 这里我将 kafka broker 传入 handler 中 bk := srv.Server().Options().Broker h := handler.GetHandler(session, vClient, uClient, bk) 复制代码</code></pre>

发布消息：

<pre><code class="hljs go copyable" lang="go">// 发送log // ... data := &broker.Message{ Header: map[string]string{ "user_id": userID, }, Body: body, } if err := h.Broker.Publish(topic, data); err != nil { log.Printf("[pub] failed: %v\n", err) } 复制代码</code></pre>

订阅消息：

<pre><code class="hljs go copyable" lang="go"> bk := srv.Server().Options().Broker // 这里订阅了一个 topic, 并提供接口处理 _, err := bk.Subscribe(topic, subLog) 复制代码</code></pre> <pre><code class="hljs go copyable" lang="go">func subLog(pub broker.Publication) error { var logPB *pb.Log // 自行解析 body 即可 if err := json.Unmarshal(pub.Message().Body, &logPB); err != nil { return err } log.Printf("[Log]: user_id: %s, Msg: %v\n", pub.Message().Header["user_id"], logPB) return nil } 复制代码</code></pre> <h3 class="heading">熔断</h3>

Micro提供了两种实现，gobreaker和hystrix，熔断是在客户端实现。

来看看 hystrix：

<pre><code class="hljs go copyable" lang="go">var ( // DefaultTimeout is how long to wait for command to complete, in milliseconds DefaultTimeout = 1000 // DefaultMaxConcurrent is how many commands of the same type can run at the same time DefaultMaxConcurrent = 10 // DefaultVolumeThreshold is the minimum number of requests needed before a circuit can be tripped due to health DefaultVolumeThreshold = 20 // DefaultSleepWindow is how long, in milliseconds, to wait after a circuit opens before testing for recovery DefaultSleepWindow = 5000 // DefaultErrorPercentThreshold causes circuits to open once the rolling measure of errors exceeds this percent of requests DefaultErrorPercentThreshold = 50 // DefaultLogger is the default logger that will be used in the Hystrix package. By default prints nothing. DefaultLogger = NoopLogger{} ) type Settings struct { Timeout time.Duration MaxConcurrentRequests int RequestVolumeThreshold uint64 SleepWindow time.Duration ErrorPercentThreshold int } 复制代码</code></pre>

若想修改参数，hystrix 没有提供全局的接口修改，这里我直接修改默认参数

<pre><code class="hljs go copyable" lang="go">// "github.com/afex/hystrix-go/hystrix" hystrix.DefaultMaxConcurrent = 100 hystrix.DefaultVolumeThreshold = 50 复制代码</code></pre>

<pre><code class="hljs go copyable" lang="go">// "github.com/micro/go-plugins/wrapper/breaker/hystrix" // 添加如下配置即可 micro.WrapClient(hystrix.NewClientWrapper() 复制代码</code></pre> <h3 class="heading">限流</h3>

ratelimit 可以在客户端做，也可以在服务端做；micro提供了两种方案：<code>juju/ratelimit</code>、<code>uber/ratelimit</code>。

我们看看 uber 的：

<pre><code class="hljs go copyable" lang="go">// "github.com/micro/go-plugins/wrapper/ratelimiter/uber" // 添加如下配置即可 // ratelimit 的配置可自行查看 API 修改 micro.WrapClient(ratelimit.NewHandlerWrapper(1024)) 复制代码</code></pre> <h3 class="heading">链路追踪</h3>

这里使用 jaeger , jaeger 提供了UI界面，端口为16686

docker-compose.yml 中已经定义

<pre><code class="hljs yaml copyable"> jaeger: image: jaegertracing/all-in-one:1.12 container_name: tracing environment: COLLECTOR_ZIPKIN_HTTP_PORT: 9411 ports: - "5775:5775/udp" - "6831:6831/udp" - "6832:6832/udp" - "5778:5778" - "16686:16686" - "14268:14268" - "9411:9411" 复制代码</code></pre> <pre><code class="hljs go copyable" lang="go">func NewJaegerTracer(serviceName, addr string) (opentracing.Tracer, io.Closer, error) { // Sample configuration for testing. Use constant sampling to sample every trace // and enable LogSpan to log every span via configured Logger. cfg := jaegercfg.Configuration{ Sampler: &jaegercfg.SamplerConfig{ Type: jaeger.SamplerTypeConst, Param: 1, }, Reporter: &jaegercfg.ReporterConfig{ LogSpans: true, BufferFlushInterval: 1 * time.Second, }, } cfg.ServiceName = serviceName // Example logger and metrics factory. Use github.com/uber/jaeger-client-go/log // and github.com/uber/jaeger-lib/metrics respectively to bind to real logging and metrics // frameworks. jLogger := &jaegerLogger{} jMetricsFactory := metrics.NullFactory metricsFactory := metrics.NullFactory metrics := jaeger.NewMetrics(metricsFactory, nil) sender, err := jaeger.NewUDPTransport(addr, 0) if err != nil { log.Logf("could not initialize jaeger sender: %s", err.Error()) return nil, nil, err } repoter := jaeger.NewRemoteReporter(sender, jaeger.ReporterOptions.Metrics(metrics)) return cfg.NewTracer( jaegercfg.Logger(jLogger), jaegercfg.Metrics(jMetricsFactory), jaegercfg.Reporter(repoter), ) } type jaegerLogger struct{} func (l *jaegerLogger) Error(msg string) { log.Logf("ERROR: %s", msg) } // Infof logs a message at info priority func (l *jaegerLogger) Infof(msg string, args ...interface{}) { log.Logf(msg, args...) } 复制代码</code></pre> <pre><code class="hljs go copyable" lang="go">// "github.com/micro/go-plugins/wrapper/trace/opentracing" opentracing.NewClientWrapper(t) 复制代码</code></pre> <h3 class="heading">API</h3>

这里结合大名鼎鼎的 HTTP Restful 框架 gin 使用

主要代码如下：

<pre><code class="hljs go copyable" lang="go"> // web 的初始化不太一样 micro -> web srv := common.GetMicroWeb(service) // ... router := gin.Default() // ... 正常 gin router 绑定操作 // 最后直接将服务 / 绑定到 gin 的router上，交给 gin 处理 srv.Handle("/", router) 复制代码</code></pre> <h2 class="heading">go-micro 详解</h2>

micro 文档：micro.mu/docs/index.…

参见另一篇 go-micro详解

<h2 class="heading">micro</h2>

参见另一篇 micro 工具箱

Golang微服务开发实践

最近更新

浏览排行