在分布式系统中，Golang编写的RPC服务经常需要处理大量跨服务请求，若不对RPC调用做流量控制，很容易导致下游服务负载过高甚至崩溃。合理的RPC调用限速可以平衡系统吞吐量和稳定性，下面介绍几种可行的实现方案。
基于令牌桶算法的RPC限速实现
令牌桶是流量控制中常用的算法，核心逻辑是以固定速率向桶中放入令牌，请求到来时先从桶中取令牌，取到则允许执行，取不到则拒绝或等待。Golang标准库的golang.org/x/time/rate包已经实现了令牌桶算法，可直接用于RPC调用限速。
以下是使用rate包实现RPC客户端调用限速的示例：
package main
import (
"context"
"fmt"
"golang.org/x/time/rate"
"time"
)
// 模拟RPC调用函数
func mockRPCCall(ctx context.Context, req string) (string, error) {
// 模拟RPC处理耗时
time.Sleep(100 * time.Millisecond)
return "response for " + req, nil
}
// 带限速的RPC调用封装
type RateLimitRPCClient struct {
limiter *rate.Limiter
}
// 初始化限速客户端，r是每秒生成的令牌数，b是桶的最大容量
func NewRateLimitRPCClient(r rate.Limit, b int) *RateLimitRPCClient {
return &RateLimitRPCClient{
limiter: rate.NewLimiter(r, b),
}
}
// 执行限速后的RPC调用
func (c *RateLimitRPCClient) Call(ctx context.Context, req string) (string, error) {
// 等待获取令牌，超时则直接返回错误
if err := c.limiter.Wait(ctx); err != nil {
return "", fmt.Errorf("rpc call rate limit exceeded: %v", err)
}
return mockRPCCall(ctx, req)
}
func main() {
// 初始化限速器：每秒允许2次调用，桶最大容量为5
client := NewRateLimitRPCClient(2, 5)
ctx := context.Background()
// 模拟10次并发RPC调用
for i := 0; i < 10; i++ {
go func(idx int) {
req := fmt.Sprintf("request_%d", idx)
resp, err := client.Call(ctx, req)
if err != nil {
fmt.Printf("req %s failed: %vn", req, err)
return
}
fmt.Printf("req %s success: %sn", req, resp)
}(i)
}
// 等待所有goroutine执行完成
time.Sleep(3 * time.Second)
}
RPC服务端中间件拦截限速
除了在客户端做限速，也可以在RPC服务端通过中间件的方式统一拦截请求，实现流量控制。这种方式不需要修改每个RPC方法的逻辑，适合全局流量管控。
以下是基于net/rpc包实现服务端限速中间件的示例：
package main
import (
"fmt"
"golang.org/x/time/rate"
"net"
"net/rpc"
"sync"
"time"
)
// 定义RPC服务结构体
type UserService struct{}
// 用户查询方法
func (u *UserService) GetUser(id int, reply *string) error {
*reply = fmt.Sprintf("user_%d", id)
return nil
}
// 限速中间件结构体
type RateLimitMiddleware struct {
limiter *rate.Limiter
next rpc.ServerCodec
}
// 重写ReadRequestHeader方法，拦截请求
func (m *RateLimitMiddleware) ReadRequestHeader(r *rpc.Request) error {
ctx := context.Background()
// 尝试获取令牌，获取不到则拒绝请求
if !m.limiter.Allow() {
return fmt.Errorf("server rate limit exceeded")
}
return m.next.ReadRequestHeader(r)
}
// 其他Codec方法直接转发
func (m *RateLimitMiddleware) ReadRequestBody(body interface{}) error {
return m.next.ReadRequestBody(body)
}
func (m *RateLimitMiddleware) WriteResponse(r *rpc.Response, body interface{}) error {
return m.next.WriteResponse(r, body)
}
func (m *RateLimitMiddleware) Close() error {
return m.next.Close()
}
func main() {
// 注册RPC服务
rpc.Register(&UserService{})
// 监听端口
listener, err := net.Listen("tcp", ":1234")
if err != nil {
fmt.Printf("listen failed: %vn", err)
return
}
defer listener.Close()
// 初始化限速器：每秒处理3个请求，桶容量3
limiter := rate.NewLimiter(3, 3)
fmt.Println("rpc server start with rate limit...")
for {
conn, err := listener.Accept()
if err != nil {
fmt.Printf("accept conn failed: %vn", err)
continue
}
// 包装连接，加入限速中间件
codec := rpc.NewServerCodec(conn)
middlewareCodec := &RateLimitMiddleware{
limiter: limiter,
next: codec,
}
go rpc.ServeCodec(middlewareCodec)
}
}
结合Context的超时限速方案
有时候我们需要针对单个RPC调用设置更灵活的限速规则，比如不同优先级的请求设置不同的超时和限速阈值，这时候可以结合context的超时控制实现。
以下是结合context和令牌桶实现分级限速的示例：
package main
import (
"context"
"fmt"
"golang.org/x/time/rate"
"time"
)
// 分级限速管理器
type PriorityRateLimiter struct {
highLimiter *rate.Limiter // 高优先级限速器
lowLimiter *rate.Limiter // 低优先级限速器
}
func NewPriorityRateLimiter(highRate, lowRate rate.Limit, highBurst, lowBurst int) *PriorityRateLimiter {
return &PriorityRateLimiter{
highLimiter: rate.NewLimiter(highRate, highBurst),
lowLimiter: rate.NewLimiter(lowRate, lowBurst),
}
}
// 根据优先级执行RPC调用
func (p *PriorityRateLimiter) CallRPC(ctx context.Context, req string, isHighPriority bool) (string, error) {
var limiter *rate.Limiter
if isHighPriority {
limiter = p.highLimiter
} else {
limiter = p.lowLimiter
}
// 设置500ms超时获取令牌
timeoutCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
defer cancel()
if err := limiter.Wait(timeoutCtx); err != nil {
return "", fmt.Errorf("rate limit wait timeout: %v", err)
}
// 模拟RPC调用
time.Sleep(200 * time.Millisecond)
return "response for " + req, nil
}
func main() {
// 高优先级每秒5次，低优先级每秒2次
limiter := NewPriorityRateLimiter(5, 2, 5, 2)
ctx := context.Background()
// 模拟高优先级请求
go func() {
resp, err := limiter.CallRPC(ctx, "high_req_1", true)
if err != nil {
fmt.Printf("high priority req failed: %vn", err)
return
}
fmt.Printf("high priority req success: %sn", resp)
}()
// 模拟低优先级请求
go func() {
resp, err := limiter.CallRPC(ctx, "low_req_1", false)
if err != nil {
fmt.Printf("low priority req failed: %vn", err)
return
}
fmt.Printf("low priority req success: %sn", resp)
}()
time.Sleep(1 * time.Second)
}
不同限速方案的选择建议
实际开发中可以根据场景选择合适的RPC限速方案：
- 如果是客户端侧需要控制自身发起的RPC请求频率，避免打爆下游服务，优先选择客户端令牌桶限速方案，实现简单且对服务端无侵入。
- 如果是服务端需要全局管控所有接入的RPC请求流量，选择服务端中间件拦截方案更合适，方便统一调整限速规则。
- 如果系统存在不同优先级的RPC请求，需要差异化限速，可以结合context和分级限速管理器实现，兼顾不同请求的处理需求。
需要注意的是，限速阈值需要根据服务的实际处理能力压测确定，避免阈值设置过高起不到保护作用，或者设置过低影响正常业务请求的处理。