Help us shape the future of Go! Take the 2026 Go Developer Survey

kitexx

package module
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 16, 2026 License: MIT Imports: 36 Imported by: 0

README

⚠️ 已改名：oekit/kitex → oekit/kitexx

本模块已更名为 oekit/kitexx（v0.4.0 起）—— 避免与 cloudwego/kitex 命名混淆，与 redisx/gormx/hertzx/logx 的 -x 系对齐。

  • 旧路径 github.com/oekit/kitexx 经 GitHub 重定向仍可拉取 ≤ v0.3.11
  • 结构：原 kitex/ 核心子包上收为根包（package kitexx），nns/registry/observability/kitexmeta 子树随新路径
go get github.com/oekit/kitexx@v0.4.0

Documentation

Overview

Package kitexx 为 CloudWeGo Kitex RPC 框架提供 ekit 生态集成。

本包覆盖以下能力：

快速开始 

// 1. 日志桥接（推荐 logx/kitexlog；必须传 slog 实例避免无限递归）
klog.SetLogger(kitexlog.NewLogger(logx.NewSlogLogger("my-rpc")))
//    或 Deprecated 别名：klog.SetLogger(kitex.NewLogger(logx.NewSlogLogger("my-rpc")))

// 2. 可观测性（需先调用 tracing.Init）
svr := echo.NewServer(new(Impl),
    server.WithSuite(kitex.NewServerTracingSuite()),
)

// 3. 治理配置（需 Nacos config client）
gov, _ := kitex.NewGovernanceConfig(configCli)
svr := echo.NewServer(new(Impl),
    server.WithSuite(kitex.NewServerSuite("echo", gov)),
)

优雅退出最佳实践（对照官方 best practice 文档）

Kitex 的 server.Stop() 按以下流程排空连接：

  1. 停止接受新连接（关闭监听端口）
  2. 非多路复用：每秒检查直到所有连接处理完毕；多路复用：先发控制帧通知客户端再等待
  3. 超时退出：达到 ExitWaitTime（默认 5s）后直接退出

关键旋钮： 

server.WithExitWaitTime(10*time.Second) // 排空等待上限（默认 5s）

时间关系（K8s 环境下必须满足）： 

client.DiscoverLatency < k8s.PreStopTime < server.MaxLatency < kitex.ExitWaitTime < k8s.TerminationGracePeriodSeconds

其中：

  • client.DiscoverLatency：客户端服务发现延迟（Nacos 推送 + 缓存刷新）
  • k8s.PreStopTime：preStop 命令等待时间（建议 sleep 5s，确保 iptables/endpoint 更新传播）
  • server.MaxLatency：单请求最大处理耗时
  • kitex.ExitWaitTime：框架排空等待上限
  • k8s.TerminationGracePeriodSeconds：K8s 强杀超时（须大于以上所有值之和）

推荐 K8s 配置： 

lifecycle:
  preStop:
    exec:
      command: ["sleep", "5"]
terminationGracePeriodSeconds: 30

与 ekit bootstrap 集成时，bootstrap 的 ShutdownTimeout 是单资源关闭预算； Kitex server.Stop() 使用自身 ExitWaitTime（忽略传入 ctx），两者独立。 保持 ShutdownTimeout > ExitWaitTime 以确保排空不被 bootstrap 兜底截断。

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func FromKitexError 

func FromKitexError(err error) (*ekiterr.BizError, bool)

FromKitexError 把 kitex kerrors 错误转为 ekit BizError,供客户端统一 错误处理路径(middleware.Fail、StatusMapper)使用。覆盖以下类型:

  • *kerrors.BizStatusError / *kerrors.GRPCBizStatusError(对端业务异常): 透传 BizStatusCode 作为 BizCode,BizMessage 作为 Message; HTTPCode 留 0,由 ekit StatusMapper 推导。
  • kerrors.IsTimeoutError(链路超时):BizCode=50400, HTTPCode=504。
  • kerrors.ErrNoMoreInstance / ErrACL(无可用实例 / ACL 拒绝): BizCode=50300 / 40301, HTTPCode=503 / 403。
  • *kerrors.DetailedError 兜底:BizCode=50000, HTTPCode=500。
  • nil / 非 kitex 错误:返回 (nil, false),调用方走 fallback。

原始错误始终保留为 BizError.Cause(经 Unwrap 链可达):errors.Is / errors.As / kerrors.FromBizStatusError 等链式判定对转换结果依然有效, DetailedError 携带的底层 cause 不会丢失(2026-09-05 复审修复:原实现 丢弃原始错误,转换后排障信息断链)。

官方依据:https://www.cloudwego.io/zh/docs/kitex/best-practice/error_handle/ ("kerrors 的错误定义支持 errors.Is 和 errors.Unwrap"、"basic error types 可以用 WithCause 来添加详细原因" — 错误链可追溯是 kitex 官方语义)

用法(典型场景:kitex client middleware,把对端错误转 ekit BizError 上抛): 

if biz, ok := kitex.FromKitexError(err); ok {
    return biz
}
return err

归属说明(2026-09-19 重构):本函数原在 pkg/errors(kerrors_bridge.go), 属"pkg/errors import kitex kerrors"明示例外;为恢复 L1 基础层零框架 依赖,迁至 kitex 包——kitex 错误桥接本就是 kitex 专属能力。

func IsKitexTimeoutError 

func IsKitexTimeoutError(err error) bool

IsKitexTimeoutError 包装 kerrors.IsTimeoutError,提供不直接 import kitex kerrors 的判定入口。

官方依据:best-practice/error_handle (Kitex 内部错误章节"判断具体错误类型")

func NewClientTracingSuite 

func NewClientTracingSuite(opts ...tracing.Option) client.Suite

NewClientTracingSuite 返回 Kitex Client 的 OpenTelemetry tracing suite。

前提：ekit tracing.Init 已在 bootstrap 阶段调用，OTel 全局 TracerProvider/Propagator 已设置。 该 Suite 会注入 tracing middleware + TTHeader/HTTP2 meta handler + transport 协议。

func NewLogger deprecated 

func NewLogger(logger logx.Logger) klog.FullLogger

NewLogger 将 ekit logx.Logger 适配为 Kitex klog.FullLogger。

映射关系（与迁移前完全一致）：

  • Trace → Debug（logx 无 Trace 级别）
  • Notice → Info（logx 无 Notice 级别）
  • Fatal → Error（ekit 不做 os.Exit）

必须传 slog 实例，不可传 hlog 后端的 Logger（如 logx.WithService(...)）， 否则会形成 adapter → logx → hlog → adapter 的无限递归；实现会在检测到 hlog 后端时 panic 以防止生产环境不可控故障。

Deprecated: 已迁至 github.com/oekit/logx/kitexlog（kitexlog.NewLogger）， 本别名委托实现，行为完全一致；将在未来 major 版本移除。

func NewPanicLogMiddleware 

func NewPanicLogMiddleware() endpoint.Middleware

NewPanicLogMiddleware 返回记录业务 handler panic 的 server middleware。

kitex 框架在 handler panic 时自动 recover 并把详情写入 rpcinfo stats (返回 kerrors.ErrPanic/*kerrors.DetailedError 给调用方),业务无法在 recover 之前插桩;官方推荐的获取详情方式就是在 middleware 里检查 rpcinfo.GetRPCInfo(ctx).Stats().Panicked()。本 middleware 做的事:

  • next 返回 error 且 stats 记录了 panic → 写 logx 结构化错误日志 (method / remote_addr / panic 值 / err / stack),err 原样透传, 不改变 kitex 默认错误返回语义;
  • 无 panic 的普通错误、无 rpcinfo 的 ctx:直接透传,零开销。

用法(与 SetDefaultPanicHandler 互补,建议一起挂): 

svr := echo.NewServer(handler,
    server.WithMiddleware(ekitkitex.NewPanicLogMiddleware()),
)

官方依据:https://www.cloudwego.io/zh/docs/kitex/tutorials/basic-feature/panic/ ("你可以在自己的中间件里检查是否发生过 panic" → ri := rpcinfo.GetRPCInfo(ctx); ri.Stats(); stats.Panicked())

func NewServerTracingSuite 

func NewServerTracingSuite(opts ...tracing.Option) server.Suite

NewServerTracingSuite 返回 Kitex Server 的 OpenTelemetry tracing suite。

前提：ekit tracing.Init 已在 bootstrap 阶段调用，OTel 全局 TracerProvider/Propagator 已设置。 该 Suite 会注入 tracing middleware + TTHeader/HTTP2 meta handler。

func RunServer 

func RunServer(app lifecycle.App, name string, svr server.Server)

RunServer 将 Kitex Server 注册到 ekit 生命周期管理（bootstrap.App 满足 lifecycle.App 接口；transport 适配层只依赖 lifecycle 契约，不依赖 bootstrap）。

该函数会：

  1. 通过 app.AddResource 注册 svr.Stop 作为关闭回调（优雅关闭时自动调用）。
  2. 通过 app.GoBackground 在后台启动 svr.Run()（阻塞直到 server 退出）。
  3. svr.Run() 返回 error 或 panic 时（netpoll 多路复用层在监听失败如端口占用时 直接 panic 而非返回 error）调用 app.RequestShutdown() 触发优雅关机， 避免进程无服务地"假活"（对称 hertz.RunServer）。

关闭语义（对照官方 best-practice/graceful_shutdown 文档）

信号路径：SIGTERM/SIGINT（SIGHUP 未被忽略时含之）由 svr.Run() 内置 waitSignal 处理，直接触发 svr.Stop(context.Background())，排空上限为 server.WithExitWaitTime（**默认 5s** — kitex 官方默认;如未显式设置,框架 内部用 5s,非多路复用每秒 ticker 检 active conn / 多路复用先发控制帧）。 同一信号也会让 bootstrap 走本函数注册的关闭回调,第二次 Stop 因 server 已关可能返回 ErrServerDone（kitex 0.16+ 行为）— 跟 hertz 双路关闭的良性 竞争同理,本函数按成功处理。

⚠️ **必须满足 K8s 时间关系**(官方文档强制约束,违反则优雅停机不保证): 

client.DiscoverLatency < k8s.PreStopTime < server.MaxLatency
    < kitex.ExitWaitTime < k8s.TerminationGracePeriodSeconds

**推荐显式设置 server.WithExitWaitTime**(尤其长耗时 RPC):保持其小于 bootstrap 单资源关闭预算(shutdown.New 的 defaultTimeout),避免排空被 bootstrap 兜底截断。kitex server.Stop() 忽略传入 ctx,使用自身 ExitWaitTime。

使用方式（在 bootstrap.Run 的 userFn 内）： 

b, _ := bootstrap.New(bootstrap.WithServiceName("svc")) // + nacos/obs WithXxx 按需注入
b.Run(func(app *bootstrap.App) error {
    svr := echo.NewServer(new(Impl),
        server.WithRegistry(kitex.NewRegistry(reg)),
        server.WithServiceAddr(addr.NewNetAddr("tcp", ":9090")),
    )
    kitex.RunServer(app, "kitex-echo", svr)

    h := server.New(server.WithHostPorts(":8080"))
    app.AddResource("hertz", func(ctx context.Context) error { return h.Shutdown(ctx) })
    return h.Run()
})

func SetDefaultPanicHandler 

func SetDefaultPanicHandler()

SetDefaultPanicHandler 注册 ekit 风格的 kitex **内部 goroutine** panic handler (gofunc.RecoverGoFuncWithInfo 恢复的 goroutine:ttstream/nphttp2 流式传输、 框架后台任务等),日志走 logx 结构化输出。

⚠️ 覆盖范围(2026-09-05 复审修正,以 kitex v0.16.3 源码为准):

  • 业务 handler panic:kitex 在 server/server.go invokeHandleEndpoint 的 defer 里直接 recover 并转 kerrors.ErrPanic,**不经过** gofunc 的 panicHandler。要在统一日志里记录 handler panic,请使用 NewPanicLogMiddleware(官方推荐路径)。
  • kitex 内部 goroutine panic(流式传输等):走 gofunc.SetPanicHandler, 本函数生效。注意这是对 kitex 默认行为的叠加——gofunc 内部仍会打 klog,本 handler 额外写 logx。

官方依据:https://www.cloudwego.io/zh/docs/kitex/tutorials/basic-feature/panic/ ("业务代码使用 go 关键字创建的 goroutine 里发生的 panic,需要业务自行 recover"——handler panic 由框架 recover,详情走 Stats().Panicked(),官方 文档未将 gofunc.SetPanicHandler 列为 handler panic 钩子)

用法(在 bootstrap.Run userFn 内,kitex.RunServer 之前): 

b, _ := bootstrap.New(bootstrap.WithServiceName("svc")) // + nacos/obs WithXxx 按需注入
b.Run(func(app *bootstrap.App) error {
    ekitkitex.SetDefaultPanicHandler()
    svr := echo.NewServer(new(Impl), ...)
    ekitkitex.RunServer(app, "kitex-echo", svr)
    ...
})

func WithBizErrorBridge 

func WithBizErrorBridge() client.Option

WithBizErrorBridge 返回 kitex client middleware,把对端抛的 kitex kerrors (BizStatusError / GRPCBizStatusError / DetailedError / timeout / no-instance / ACL) 转为 ekit BizError 沿原 error 通道上抛,让 ekit handler 层 middleware.Fail + StatusMapper 走统一处理路径,不再 fallback 500。

⚠️ 行为变化:开启后,client 调用的 error 返回类型可能从 kitex kerrors 变成 ekit *BizError,使用 errors.Is/As 判定的代码需注意。pkg/errors.AsBizError 仍可识别(*BizError Unwrap 自身 cause,kerrors.BizStatusError 通过 kerrors.FromBizStatusError 提取)。

官方依据: 

https://www.cloudwego.io/zh/docs/kitex/best-practice/error_handle/
https://www.cloudwego.io/zh/docs/kitex/tutorials/basic-feature/bizstatuserr/

用法(对称其他 client.Option): 

cli, err := myservice.NewClient("myservice",
    kitex.WithBizErrorBridge(),
    client.WithHostPorts(":8888"),
)

⚠️ kitex v0.16.3 实测(e2e 2026-09-06):普通 thrift/TTHeader ping-pong 下, server 端 BizStatusError 不做类型化解码(仅 ttstream/gRPC 流式支持), 到达 client 的是 DetailedError 文本 → 走 50000 兜底分支;类型化 code/msg 映射需要流式协议。详见 kitex/e2e_meta_test.go。

Types

type ClientSuite 

type ClientSuite struct {
	// contains filtered or unexported fields
}

ClientSuite 聚合 Kitex 客户端治理选项（retry + rpc_timeout + circuit_breaker）。

实现 client.Suite 接口，可通过 client.WithSuite 一行接入。

func NewClientSuite 

func NewClientSuite(clientService, serverService string, gov *GovernanceConfig, opts ...ClientSuiteOption) *ClientSuite

NewClientSuite 创建客户端治理 Suite。

clientService 为当前服务名，serverService 为被调服务名。 governance 为 NewGovernanceConfig 返回的治理配置客户端。

使用方式： 

suite := kitex.NewClientSuite("my-service", "echo-service", gov,
    kitex.WithRetry(),
    kitex.WithRPCTimeout(),
)
cli, _ := echo.NewClient("echo-service", client.WithSuite(suite))

func (*ClientSuite) Close 

func (s *ClientSuite) Close() error

Close 释放本 Suite 创建、且不受 kitex client 生命周期管理的治理资源 （retry 容器内部的熔断面板；CBSuite 由 client.Close() 自动关闭，无需在此处理）。

必须在所有由本 Suite 构建的 client 关闭之后再调用——提前调用会使 retry 容器的重试熔断面板失效。幂等，可安全重复调用。

func (*ClientSuite) Options 

func (s *ClientSuite) Options() []client.Option

Options 返回聚合的 client.Option 列表。

type ClientSuiteOption 

type ClientSuiteOption func(*clientSuiteConfig)

ClientSuiteOption 是 NewClientSuite 的配置选项函数类型。

func WithCircuitBreaker 

func WithCircuitBreaker() ClientSuiteOption

WithCircuitBreaker 启用 circuit_breaker 策略热更新。

func WithMetrics 

func WithMetrics(reg prom.Registerer, namespace string) ClientSuiteOption

WithMetrics 启用弹性指标（熔断状态/迁移/拒绝/重试，kind="kitex"）， 形态对齐 httpx.WithMetrics 与 nns endpoint.WithMetrics。 reg 为 nil 时无指标；namespace 为空时用 "kitex"。 重试计数经 metainfo 的 retry.TransitKey（kitex retry 容器每次重试 重新走 middleware 链并注入序号）；方法级 RPC 视角（吞吐/时延按 caller/callee/method/status/retry 分维）另见 kitexprom 桥接包。

func WithRPCTimeout 

func WithRPCTimeout() ClientSuiteOption

WithRPCTimeout 启用 rpc_timeout 策略热更新。

func WithRetry 

func WithRetry() ClientSuiteOption

WithRetry 启用 retry 策略热更新。

type DNSResolver 

type DNSResolver = discovery.Resolver

DNSResolver 是 kitex-contrib/resolver-dns 的别名 type,kitex 客户端 把它当 discovery.Resolver 用(供 client.WithResolver(option) 接入)。

官方依据:https://github.com/kitex-contrib/resolver-dns (v0.1.0)

业务场景: K8s/容器编排下,服务注册用 Headless Service / ClusterIP 暴露,业务用 DNS 名做服务发现,ekit 把 DNSResolver 喂给 kitex client 替代静态 IP 列表。

业务用法: 

cli, _ := myservice.NewClient("myservice",
    client.WithResolver(mw.NewDNSResolver()),
    ...
)

注意: DNS lookup 在 K8s 内网/Headless Service 下,每个 pod 有 独立 IP(轮询 round-robin),适合 stateless 微服务;stateful 服务(有状态)用 Headless Service 配 hostname 替代。

func NewDNSResolver 

func NewDNSResolver() DNSResolver

NewDNSResolver 桥接 kitex-contrib/resolver-dns.NewDNSResolver。 业务用 client.WithResolver(ekit.NewDNSResolver()) 即可。

局限: kitex-contrib/resolver-dns 只解析 A/AAAA 记录(SRV/Service 头解析需要业务自己实现),如需 SRV 记录可换 kitex-contrib/registry-zookeeper 或 ekit 内部 NNS。

type GovernanceConfig 

type GovernanceConfig struct {
	// contains filtered or unexported fields
}

GovernanceConfig 从 Nacos 配置中心加载 Kitex 治理策略（retry/rpc_timeout/circuit_breaker/limiter） 并在配置变更时热更新到 Kitex 运行时。

每类策略对应一个独立的 Nacos 配置项（dataId），格式为： 

客户端策略: {clientService}.{serverService}.{category}
服务端策略: {serverService}.{category}

category 可选值: retry, rpc_timeout, circuit_break, limit, degradation

func NewGovernanceConfig 

func NewGovernanceConfig(cli config_client.IConfigClient, opts ...GovernanceOption) (*GovernanceConfig, error)

NewGovernanceConfig 创建治理配置客户端。

cli 为 ekit [nacos.NewConfigClient] 返回的配置客户端，与业务配置共享同一连接。

func (*GovernanceConfig) Close 

func (g *GovernanceConfig) Close() error

Close 取消本 GovernanceConfig 注册的全部 Nacos 监听并清空订阅状态。

动态创建依赖 client 的场景（如网关按需建 client）会随时间累积监听器， 在 client 全部关闭后调用 Close 释放治理订阅。Close 后对象仍可复用： 下一次 subscribe 会重新注册监听，但未重新订阅前不再收到任何回调。

注意：nacos 的 CancelListenConfig 以 dataId 为粒度，会一并移除该 dataId 上其他监听者——请勿对同一 dataId 混用不同 GovernanceConfig 实例。

func (*GovernanceConfig) TemplateTimeoutOptions 

func (g *GovernanceConfig) TemplateTimeoutOptions() []client.Option

TemplateTimeoutOptions 返回统一模板 timeout_ms 的 client.Option; 未启用 WithGovernanceTemplate 时返回 nil。

type GovernanceOption 

type GovernanceOption func(*GovernanceConfig)

GovernanceOption 是 NewGovernanceConfig 的配置选项函数类型。

func WithDataIDTemplate 

func WithDataIDTemplate(tmpl string) GovernanceOption

func WithGovernanceGroup 

func WithGovernanceGroup(group string) GovernanceOption

WithGovernanceGroup 设置 Nacos 配置分组，默认 "DEFAULT_GROUP"。

func WithGovernanceTemplate 

func WithGovernanceTemplate(dataID string) GovernanceOption

WithDataIDTemplate 设置 dataId 模板，支持占位符 {client}、{server}、{category}。 默认 "{client}.{server}.{category}"（客户端）和 "{server}.{category}"（服务端）。 传入的模板同时用于客户端；服务端模板会自动去掉 "{client}." 前缀。 WithGovernanceTemplate 订阅**统一治理模板**(governance 包 schema, audit 5.2 阶段 2):单 dataID 承载全部类别,template 中 timeout_ms 应用到 rpctimeout 容器(经 TemplateTimeoutOptions 取 client.Option);retry/ circuit_breaker/limiter 为 kitex 原生富结构,仍走 per-category dataID。

type ServerSuite 

type ServerSuite struct {
	// contains filtered or unexported fields
}

ServerSuite 聚合 Kitex 服务端治理选项（limiter）。

实现 server.Suite 接口，可通过 server.WithSuite 一行接入。

func NewServerSuite 

func NewServerSuite(serverService string, gov *GovernanceConfig, opts ...ServerSuiteOption) *ServerSuite

NewServerSuite 创建服务端治理 Suite。

serverService 为当前服务名。 governance 为 NewGovernanceConfig 返回的治理配置客户端。

使用方式： 

suite := kitex.NewServerSuite("echo-service", gov, kitex.WithLimiter())
svr := echo.NewServer(new(Impl), server.WithSuite(suite))

func (*ServerSuite) Options 

func (s *ServerSuite) Options() []server.Option

Options 返回聚合的 server.Option 列表。

type ServerSuiteOption 

type ServerSuiteOption func(*serverSuiteConfig)

ServerSuiteOption 是 NewServerSuite 的配置选项函数类型。

func WithLimitReporter 

func WithLimitReporter(r klimiter.LimitReporter) ServerSuiteOption

WithLimitReporter 注入限流过载上报器（kitex 在连接/QPS 超限拒绝请求时回调）。 未注入时默认使用内置的节流日志上报（每类事件至多每秒一条 Warn）。

func WithLimiter 

func WithLimiter() ServerSuiteOption

WithLimiter 启用 limiter 策略热更新。

Source Files

View all Source files

Directories

Path Synopsis
internal
e2e/kitex_gen/kitex/echo
e2e/kitex_gen/kitex/echo/echoservice
Code generated by Kitex v0.16.3.
 Code generated by Kitex v0.16.3.
Package kitexmeta 桥接 kitex transmeta 体系 + ekit traceid,让 trace_id / 用户身份 / 空间码等业务元数据在 kitex server 入口注入 到 ctx,通过 kitex client transmeta 自动透传到下游 server。
 Package kitexmeta 桥接 kitex transmeta 体系 + ekit traceid,让 trace_id / 用户身份 / 空间码等业务元数据在 kitex server 入口注入 到 ctx,通过 kitex client transmeta 自动透传到下游 server。
nns
kitex
Package kitex 提供 NNS 门面到 CloudWeGo Kitex 框架的适配器。
 Package kitex 提供 NNS 门面到 CloudWeGo Kitex 框架的适配器。
observability
metrics/instrumentation/kitexprom
Package kitexprom 桥接 kitex-contrib/monitor-prometheus，为 Kitex 提供 方法级 RPC 指标（吞吐/时延，labels 含 caller/callee/method/status/retry）。
 Package kitexprom 桥接 kitex-contrib/monitor-prometheus，为 Kitex 提供 方法级 RPC 指标（吞吐/时延，labels 含 caller/callee/method/status/retry）。
tracing/instrumentation/kitexotel
Package kitexotel 提供 Kitex 客户端/服务端的 metainfo propagation 启用 helper,以及 OTel 链路透传(通过 metainfo carrier)选项。
 Package kitexotel 提供 Kitex 客户端/服务端的 metainfo propagation 启用 helper,以及 OTel 链路透传(通过 metainfo carrier)选项。
registry
kitex
Package kitex 提供 ekit registry 抽象到 CloudWeGo Kitex 框架的适配器。
 Package kitex 提供 ekit registry 抽象到 CloudWeGo Kitex 框架的适配器。

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.