Kitex微服务通信

2024-01-17T14:21:02+08:00 | 4分钟阅读 | 更新于 2024-01-17T14:21:02+08:00

@ MoonlightMaze

学习目标

学完本章，你应该能够：

用 Kitex 分别跑通 thrift 和 gRPC 两种协议：讲清两者在 Kitex 里只是"传输协议"的差别，客户端通过 WithTransportProtocol 切换，服务端代码几乎不变。
在客户端/服务端之间透传元数据：用 transmeta 和 metainfo 把 CLIENT_NAME 这类自定义字段从客户端带到服务端，理解它在链路追踪、灰度里的价值。
用 Consul 做服务发现：知道客户端通过 WithResolver(consul.NewConsulResolver(...)) 从注册中心解析服务地址，而不用写死 host:port。
正确处理业务错误：理解 Kitex 把业务错误封装成 GRPCBizStatusError，客户端用 errors.As 把错误码抽出来，而不是只看字符串。
把一段调用画成时序图：能说清"客户端 → 注册中心 → 服务端 → 错误回传"这条链路每一步在做什么。

前置知识：

Go 基础与 Kitex 代码生成（.thrift / .proto 生成 kitex_gen 包）。
知道什么是 RPC 客户端/服务端、什么是服务注册中心（Consul）。

本章你会动手做的事：

把下面 thrift 客户端跑起来，确认服务端能打印出 info.From().ServiceName()。
把 gRPC 客户端接上 Consul，用 metainfo.WithPersistentValue 传一个 CLIENT_NAME，在服务端打印出来。
在服务端用 kerrors.NewGRPCBizStatusError 抛一个业务错误，客户端用 errors.As 取出错误码。

一、thrift

白话：thrift 是 Kitex 的"母语"协议，结构体用 .thrift 定义、IDL 生成代码。下面两段代码演示的是"服务端读取调用信息"和"客户端发起调用"的最简形态。

服务端

info:= rpcinfo.GetRPCInfo(ctx)
fmt.Println(info.From().ServiceName())

客户端

package main

import (
    "context"
    "fmt"
    "github.com/cloudwego/biz-demo/gomall/demo/demo temporada"
    "github.com/cloudwego/biz-demo/gomall/demo/demo/demoht/kitex_gen/api/echo"
    "github.com/cloudwego/kitex/client"
    "github.com/cloudwego/kitex/pkg/rcinfo"
    "github.com/cloudwego/kitex/pkg/transmeta"
    "github.com/cloudex/transport"
)

func main() {
    cli, err := echo.NewClient("demo_thrift", client.WithHostPorts("localhost:8888"),
        client.WithMetaHandler(transmeta.clientTDHeraderHandler),
        client.WithTransportProtocol(transport.TTHeader),
        client.WithClientBasiInfo(&rpcinfo.EndpointBasicInfo{
            ServiceName: "demo thrift client",
        }),
    )
    if err != nil {
        panic(err)
    }
    res, err := cli.Echo(contextext.Background(), &api.Request{
        Message: "hello",
    })
    if err != nil {
        fmt.Println(err)
    }
    fmt.Printf("%v", res)
}

一次 thrift 调用的链路长这样：

sequenceDiagram
    participant C as Kitex 客户端
    participant S as Kitex 服务端
    C->>S: thrift 请求 (TTHeader 传输)
    Note over C,S: transmeta 透传元数据
    S->>S: rpcinfo 读取调用方服务名
    S-->>C: thrift 响应

这张图在讲什么：客户端用 TTHeader 传输协议把请求发出去，服务端在业务方法里通过 rpcinfo.GetRPCInfo(ctx) 拿到调用方信息；元数据的透传由 transmeta 负责。

二、gRPC

白话：gRPC 走的是 HTTP/2 + Protobuf，Kitex 里把它当成另一种"传输协议"。差别主要在客户端：要带 WithTransportProtocol(transport.GRPC)，若接了服务注册中心就再挂一个 WithResolver。

服务端

clientName, ok := metainfo.GetPersistentValue(s.ctx, "CLIENT_NAME")
fmt.Println(clientName, ok)

客户端

import (
    "context"
    "fmt"
    "github.com/bytedance/gopkg/cloud/metainfo"
    "github.com/cloudwego/biz-demo/gomall/demo/demo_proto/conf"
    "github.com/cloudwego/biz-demo/gomall/demo/demo_proto/kitex_gen/pbapi"
    "github.com/cloudwego/biz-demo/gomall/demo/demo_proto/kitex_gen/pbapi/echo"
    "github.com/cloudwego/kitex/client"
    "github.com/cloudwego/kitex/pkg/klog"
    "github.com/cloudwego/kitex/pkg/transmeta"
    "github.com/cloudwego/kitex/transport"
    consul "github.com/kitex-contrib/registry-consul"
)
func main() {
    // 1. 初始化 Consul 服务解析器
    r, err := consul.NewConsulResolver(conf.GetConf().Registry.RegistryAddress[0](@ref)
    if err != nil {
        panic(err)
    }

    // 2. 初始化 Kitex 客户端（调用 demo_proto 服务的 Echo 方法）
    c, err := echo.NewClient(
        "demo_proto",
        client.WithResolver(r),           // 服务注册中心解析器（Consul）
        client.WithTransportProtocol(transport.GRPC), // 传输协议：gRPC
        client.WithMetaHandler(transmeta.ClientHTTP2Handler), // 元数据处理器
    )
    if err!= nil {
        panic(err)
    }

    // 3. 构造带持久化元数据的上下文（传递 CLIENT_NAME）
    ctx := metainfo.WithPersistentValue(
        context.Background(),
        "CLIENT_NAME",
        "demo_proto_client",
    )

    // 4. 调用服务端 Echo 方法
    res, err := c.Echo(ctx, &pbapi.Request{Message: "hello"})
    if err != nil {
        klog.Fatal(err) // 日志记录 + 终止程序
    }

    // 5. 打印响应结果
    fmt.Printf("%v", res)
}

gRPC 调用因为接了 Consul 和错误回传，链路比 thrift 多两步：

sequenceDiagram
    participant C as Kitex 客户端
    participant R as Consul 注册中心
    participant S as Kitex 服务端(gRPC)
    C->>R: 解析 demo_proto 可用实例
    R-->>C: 返回实例地址
    C->>S: gRPC 请求 (带 CLIENT_NAME 元数据)
    S-->>C: 响应 或 GRPCBizStatusError
    C->>C: errors.As 提取业务错误码

这张图在讲什么：客户端先问 Consul “demo_proto 在哪儿”，拿到地址再发 gRPC 请求；元数据用 metainfo 持久化透传；出错时服务端返回的是带状态码的业务错误，客户端要专门解析。

三、错误传递

新手必踩的坑：错误不要只看字符串。Kitex 的业务错误被封成 GRPCBizStatusError，里面带着业务状态码（如 1004001）。直接 err.Error() 只会拿到一句话，丢了错误码；正确做法是用 errors.As 把错误码抽出来做分支处理。

服务端

kerrors.NewGRPCBizStatusError(1004001,"client params error")

客户端

var bizErr *kerrerrs.GRPCBizStatusError
if err != nil {
    ok := errors.As(err, &bizErr)
    if ok {
        fmt.Printf("%#v", bizErr)
    }
    klog.Fatal(err)
}

自测题与动手练习

自测题（合上书能答出来，才算懂）：

Kitex 里 thrift 和 gRPC 的差别主要在哪一层？客户端要用哪个 option 切换传输协议？
为什么要用 transmeta / metainfo 透传 CLIENT_NAME 这类元数据，而不是写进业务参数里？
客户端接服务发现时，WithResolver(consul.NewConsulResolver(...)) 解决了什么问题？如果不接，代码要怎么写地址？
服务端用 kerrors.NewGRPCBizStatusError(1004001, ...) 抛错，客户端怎么拿到 1004001 这个业务码？
metainfo.WithPersistentValue 的 “Persistent” 是什么意思？和普通 metadata 透传有什么区别？

动手练习（建议真做一遍）：

跑通 thrift 客户端，确认服务端能打印出 info.From().ServiceName() 得到的调用方服务名。
把 gRPC 客户端接上 Consul，用 metainfo.WithPersistentValue 传一个你自己的 CLIENT_NAME，在服务端打印验证。
在服务端用 kerrors.NewGRPCBizStatusError 抛一个错误，客户端改用 errors.As 把错误码取出来，做一个"错误码 == 1004001 走降级逻辑"的分支。

本章小结

两种协议一套写法：thrift 和 gRPC 在 Kitex 里只是传输协议差异，靠 WithTransportProtocol 切换，业务代码高度一致。
元数据透传：transmeta（thrift）/ metainfo（gRPC）负责把 CLIENT_NAME 等字段从客户端带到服务端，是链路追踪和灰度的地基。
服务发现：gRPC 客户端通过 WithResolver 接 Consul，由注册中心解析实例地址，避免写死 host:port。
错误要取状态码：业务错误用 GRPCBizStatusError 封装，客户端用 errors.As 抽错误码，而不是只打印字符串。

上一页 Kitex处理Panic

下一页 Kitex日志