rpcx 框架文档

什么是 rpcx

rpcx 是一个分布式的 Go 语言 RPC 框架，支持多种协议、多种注册中心、多种负载均衡策略。它简单易用、性能优越，是构建微服务的理想选择。

核心优势

快速开始

安装

go get -u github.com/smallnest/rpcx/...

定义服务

首先定义服务的参数和返回值结构，以及服务方法：

package service

type Args struct {
    A int
    B int
}

type Reply struct {
    C int
}

type Arith int

func (t *Arith) Mul(ctx context.Context, args *Args, reply *Reply) error {
    reply.C = args.A * args.B
    return nil
}

服务端

package main

import (
    "context"
    "github.com/smallnest/rpcx/server"
)

func main() {
    s := server.NewServer()
    s.RegisterName("Arith", new(Arith), "")
    s.Serve("tcp", "localhost:8972")
}

客户端

package main

import (
    "context"
    "github.com/smallnest/rpcx/client"
)

func main() {
    d, _ := client.NewPeer2PeerDiscovery("tcp@localhost:8972", "")
    xclient := client.NewXClient("Arith", client.Failtry, client.RandomSelect, d, client.DefaultOption)
    defer xclient.Close()

    args := &Args{A: 10, B: 20}
    reply := &Reply{}
    err := xclient.Call(context.Background(), "Mul", args, reply)

    // reply.C = 200
}

核心特性总览

rpcx 提供了丰富的企业级 RPC 特性，以下按功能分类：

服务发现与注册

负载均衡策略

故障处理模式

服务发现

服务发现是微服务架构的核心组件。rpcx 支持多种主流注册中心。

使用 etcd 作为注册中心

服务端：

import "github.com/rpcxio/rpcx-etcd/serverplugin"

s := server.NewServer()

r := &serverplugin.EtcdV3RegisterPlugin{
    ServiceAddress: "tcp@localhost:8972",
    EtcdServers:    []string{"localhost:2379"},
    BasePath:       "/rpcx_test",
    UpdateInterval: time.Minute,
}
err := r.Start()
if err != nil {
    log.Fatal(err)
}
s.Plugins.Add(r)

客户端：

d := client.NewEtcdV3Discovery("/rpcx_test", "Arith", []string{"localhost:2379"}, nil)
xclient := client.NewXClient("Arith", client.Failtry, client.RandomSelect, d, client.DefaultOption)

服务元数据

可以通过元数据对服务进行分组、设置状态等：

// 设置服务分组
s.RegisterName("Arith", new(Arith), "group=test")

// 设置服务状态（可用于灰度发布）
s.RegisterName("Arith", new(Arith), "state=active")

负载均衡

rpcx 提供多种负载均衡策略，可根据业务场景选择。

加权轮询

// 定义服务器的权重
servers := []*client.KVPair{
    {Key: "tcp@localhost:8972", Value: "weight=7"},  // 70% 流量
    {Key: "tcp@localhost:8973", Value: "weight=3"},  // 30% 流量
}

d, _ := client.NewMultipleServersDiscovery(servers)
xclient := client.NewXClient("Arith", client.Failtry, client.WeightedRoundRobin, d, client.DefaultOption)

一致性哈希

一致性哈希确保相同参数的请求始终路由到同一服务实例：

d, _ := client.NewPeer2PeerDiscovery("tcp@localhost:8972", "")
xclient := client.NewXClient("Arith", client.Failtry, client.ConsistentHash, d, client.DefaultOption)

故障处理

rpcx 提供多种故障处理模式，确保服务的高可用性。

故障转移 (Failover)

当调用失败时，自动尝试其他服务实例：

d, _ := client.NewMultipleServersDiscovery([]*client.KVPair{
    {Key: "tcp@localhost:8972"},
    {Key: "tcp@localhost:9981"},
})

option := client.DefaultOption
option.Retries = 10  // 最大重试次数

xclient := client.NewXClient("Arith", client.Failover, client.RandomSelect, d, option)

超时控制

ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()

err := xclient.Call(ctx, "Mul", args, reply)

熔断器

当服务实例连续失败达到阈值时，熔断器会暂时切断对该实例的调用，快速失败。

option := client.DefaultOption

// 连续失败 5 次后熔断，30 秒后尝试恢复
option.GenBreaker = func() client.Breaker {
    return client.NewConsecCircuitBreaker(5, 30*time.Second)
}

d, _ := client.NewMultipleServersDiscovery(servers)
xclient := client.NewXClient("Arith", client.Failfast, client.RandomSelect, d, option)

编码与序列化

rpcx 支持多种序列化方式，可根据需求选择。

使用 JSON 序列化

option := client.DefaultOption
option.SerializeType = protocol.JSON

xclient := client.NewXClient("Arith", client.Failtry, client.RandomSelect, d, option)

数据压缩

option := client.DefaultOption
option.CompressType = protocol.Gzip  // 启用 gzip 压缩

传输协议

rpcx 支持多种传输协议，适应不同场景需求。

s.Serve("tcp", "localhost:8972")

// go run -tags quic
s.Serve("quic", "localhost:8972")

// go run -tags kcp
s.Serve("kcp", "localhost:8972")

s.Serve("ws", "localhost:8972")

s.Serve("unix", "/tmp/rpcx.socket")

安全与认证

Token 认证

服务端：

s := server.NewServer()

// 设置认证函数
s.AuthFunc = func(ctx context.Context, meta string) error {
    if meta != "valid-token" {
        return errors.New("invalid token")
    }
    return nil
}

客户端：

d, _ := client.NewPeer2PeerDiscovery("tcp@localhost:8972", "")
xclient := client.NewXClient("Arith", client.Failtry, client.RandomSelect, d, client.DefaultOption)

// 设置认证 token
xclient.Auth("valid-token")

TLS 加密

// 服务端
s := server.NewServer(server.WithTLSConfig(tlsConfig))

// 客户端
option.TLSConfig = tlsConfig
xclient := client.NewXClient("Arith", client.Failtry, client.RandomSelect, d, option)

IP 黑名单

import "github.com/smallnest/rpcx/serverplugin"

plugin := serverplugin.NewBlacklistPlugin([]string{
    "192.168.1.100",
    "10.0.0.50",
})
s.Plugins.Add(plugin)

流式传输

rpcx 支持双向流式数据传输，适用于大文件传输、实时数据流等场景。

服务端

func streamHandler(conn net.Conn, args *share.StreamServiceArgs) {
    fmt.Printf("Stream connected: %v\n", args.Meta)

    // 处理流数据
    io.Copy(conn, conn)
    conn.Close()
}

func main() {
    s := server.NewServer()

    p := server.NewStreamService("localhost:8973", streamHandler, nil, 1000)
    s.EnableStreamService(share.StreamServiceName, p)

    s.Serve("tcp", "localhost:8972")
}

文件传输

rpcx 还提供了专门的文件传输服务：

p := server.NewFileTransfer("localhost:8973", uploadPath, nil, 1000)
s.EnableFileTransfer(share.FileTransferServiceName, p)

元数据传递

rpcx 支持在调用过程中传递元数据，可用于链路追踪、业务上下文传递等。

客户端发送元数据

import "github.com/smallnest/rpcx/share"

args := &Args{A: 10, B: 20}
reply := &Reply{}

// 设置请求元数据
ctx := context.WithValue(context.Background(), share.ReqMetaDataKey, map[string]string{
    "trace-id": "12345",
    "user-id":  "67890",
})

// 准备接收响应元数据
ctx = context.WithValue(ctx, share.ResMetaDataKey, make(map[string]string))

err := xclient.Call(ctx, "Mul", args, reply)

// 读取响应元数据
resMeta := ctx.Value(share.ResMetaDataKey).(map[string]string)

服务端处理元数据

func (t *Arith) Mul(ctx context.Context, args *Args, reply *Reply) error {
    // 读取请求元数据
    reqMeta := ctx.Value(share.ReqMetaDataKey).(map[string]string)

    // 写入响应元数据
    resMeta := ctx.Value(share.ResMetaDataKey).(map[string]string)
    resMeta["server"] = "rpcx-server"

    reply.C = args.A * args.B
    return nil
}

插件系统

rpcx 提供了强大的插件系统，可以在服务调用生命周期的各个阶段插入自定义逻辑。

连接插件

监控客户端连接的建立和关闭：

type ConnectionPlugin struct {}

func (p *ConnectionPlugin) HandleConnAccept(conn net.Conn) (net.Conn, bool) {
    log.Printf("client %v connected", conn.RemoteAddr().String())
    return conn, true
}

func (p *ConnectionPlugin) HandleConnClose(conn net.Conn) bool {
    log.Printf("client %v closed", conn.RemoteAddr().String())
    return true
}

// 注册插件
s.Plugins.Add(&ConnectionPlugin{})

调用包装插件

在方法调用前后执行自定义逻辑：

type WrapCall struct {}

func (p *WrapCall) PreCall(ctx context.Context, serviceName, methodName string, args interface{}) (interface{}, error) {
    log.Printf("before %s.%s: args: %v", serviceName, methodName, args)
    // 可以修改参数
    return args, nil
}

func (p *WrapCall) PostCall(ctx context.Context, serviceName, methodName string, args, reply interface{}) (interface{}, error) {
    log.Printf("after %s.%s: reply: %v", serviceName, methodName, reply)
    // 可以修改返回值
    return reply, nil
}

s.Plugins.Add(&WrapCall{})

异步调用

rpcx 支持异步调用方式，通过 Go 方法发起调用并立即返回，通过 channel 获取结果。

args := &Args{A: 10, B: 20}
reply := &Reply{}

// 发起异步调用
call, err := xclient.Go(context.Background(), "Mul", args, reply, nil)
if err != nil {
    log.Fatal(err)
}

// 等待调用完成
replyCall := <-call.Done
if replyCall.Error != nil {
    log.Fatal(replyCall.Error)
}

log.Printf("%d * %d = %d", args.A, args.B, reply.C)

异步调用适用于需要同时发起多个 RPC 调用的场景，可以显著提高并发性能。

双向通信

rpcx 支持服务端主动向客户端发送消息，实现双向通信。

服务端发送消息

var clientConn net.Conn

func (t *Arith) Mul(ctx context.Context, args *Args, reply *Reply) error {
    // 获取客户端连接
    clientConn = ctx.Value(server.RemoteConnContextKey).(net.Conn)
    reply.C = args.A * args.B
    return nil
}

// 在 goroutine 中主动推送消息
go func() {
    for clientConn != nil {
        err := s.SendMessage(clientConn, "test_service", "test_method", nil, []byte("hello"))
        if err != nil {
            break
        }
        time.Sleep(time.Second)
    }
}()

指标监控

rpcx 支持通过插件集成指标监控，可导出到 Graphite、Prometheus 等监控系统。

Metrics 插件

import "github.com/smallnest/rpcx/serverplugin"

// 创建 metrics 插件
plugin := serverplugin.NewMetricsPlugin("rpcx-server", "localhost:2003", time.Second)
s.Plugins.Add(plugin)

链路追踪

rpcx 支持 OpenTelemetry 分布式链路追踪。

OpenTelemetry 集成

import "github.com/smallnest/rpcx/protocol"
    "github.com/rpcxio/rpcx-plugins/rotel"

// 服务端
plugin := rotel.NewOpenTelemetryPlugin("rpcx-server", tracerProvider)
s.Plugins.Add(plugin)

// 客户端
option := client.DefaultOption
option.TracerPlugin = rotel.NewClientTracer("rpcx-client", tracerProvider)

调试追踪

import "github.com/smallnest/rpcx/share"

// 启用追踪日志
share.Trace = true

更多示例

rpcx 提供了丰富的示例代码，涵盖了框架的各种特性：

更多示例请参考：rpcx-examples

参考资料

• rpcx GitHub 仓库
• rpcx 示例代码
• rpcx 官方网站