4.3 玩转RPC

在不同的场景中RPC有着不同的需求,因此开源的社区就诞生了各种RPC框架。本节我们将尝试Go内置RPC框架在一些比较特殊场景的用法。

4.3.1 客户端RPC的实现原理

Go语言的RPC库最简单的使用方式是通过Client.Call方法进行同步阻塞调用,该方法的实现如下:

  1. func (client *Client) Call(
  2. serviceMethod string, args interface{},
  3. reply interface{},
  4. ) error {
  5. call := <-client.Go(serviceMethod, args, reply, make(chan *Call, 1)).Done
  6. return call.Error
  7. }

首先通过Client.Go方法进行一次异步调用,返回一个表示这次调用的Call结构体。然后等待Call结构体的Done管道返回调用结果。

我们也可以通过Client.Go方法异步调用前面的HelloService服务:

  1. func doClientWork(client *rpc.Client) {
  2. helloCall := client.Go("HelloService.Hello", "hello", new(string), nil)
  3. // do some thing
  4. helloCall = <-helloCall.Done
  5. if err := helloCall.Error; err != nil {
  6. log.Fatal(err)
  7. }
  8. args := helloCall.Args.(string)
  9. reply := helloCall.Reply.(string)
  10. fmt.Println(args, reply)
  11. }

在异步调用命令发出后,一般会执行其他的任务,因此异步调用的输入参数和返回值可以通过返回的Call变量进行获取。

执行异步调用的Client.Go方法实现如下:

  1. func (client *Client) Go(
  2. serviceMethod string, args interface{},
  3. reply interface{},
  4. done chan *Call,
  5. ) *Call {
  6. call := new(Call)
  7. call.ServiceMethod = serviceMethod
  8. call.Args = args
  9. call.Reply = reply
  10. call.Done = make(chan *Call, 10) // buffered.
  11. client.send(call)
  12. return call
  13. }

首先是构造一个表示当前调用的call变量,然后通过client.send将call的完整参数发送到RPC框架。client.send方法调用是线程安全的,因此可以从多个Goroutine同时向同一个RPC链接发送调用指令。

当调用完成或者发生错误时,将调用call.done方法通知完成:

  1. func (call *Call) done() {
  2. select {
  3. case call.Done <- call:
  4. // ok
  5. default:
  6. // We don't want to block here. It is the caller's responsibility to make
  7. // sure the channel has enough buffer space. See comment in Go().
  8. }
  9. }

Call.done方法的实现可以得知call.Done管道会将处理后的call返回。

4.3.2 基于RPC实现Watch功能

在很多系统中都提供了Watch监视功能的接口,当系统满足某种条件时Watch方法返回监控的结果。在这里我们可以尝试通过RPC框架实现一个基本的Watch功能。如前文所描述,因为client.send是线程安全的,我们也可以通过在不同的Goroutine中同时并发阻塞调用RPC方法。通过在一个独立的Goroutine中调用Watch函数进行监控。

为了便于演示,我们计划通过RPC构造一个简单的内存KV数据库。首先定义服务如下:

  1. type KVStoreService struct {
  2. m map[string]string
  3. filter map[string]func(key string)
  4. mu sync.Mutex
  5. }
  6. func NewKVStoreService() *KVStoreService {
  7. return &KVStoreService{
  8. m: make(map[string]string),
  9. filter: make(map[string]func(key string)),
  10. }
  11. }

其中m成员是一个map类型,用于存储KV数据。filter成员对应每个Watch调用时定义的过滤器函数列表。而mu成员为互斥锁,用于在多个Goroutine访问或修改时对其它成员提供保护。

然后就是Get和Set方法:

  1. func (p *KVStoreService) Get(key string, value *string) error {
  2. p.mu.Lock()
  3. defer p.mu.Unlock()
  4. if v, ok := p.m[key]; ok {
  5. *value = v
  6. return nil
  7. }
  8. return fmt.Errorf("not found")
  9. }
  10. func (p *KVStoreService) Set(kv [2]string, reply *struct{}) error {
  11. p.mu.Lock()
  12. defer p.mu.Unlock()
  13. key, value := kv[0], kv[1]
  14. if oldValue := p.m[key]; oldValue != value {
  15. for _, fn := range p.filter {
  16. fn(key)
  17. }
  18. }
  19. p.m[key] = value
  20. return nil
  21. }

在Set方法中,输入参数是key和value组成的数组,用一个匿名的空结构体表示忽略了输出参数。当修改某个key对应的值时会调用每一个过滤器函数。

而过滤器列表在Watch方法中提供:

  1. func (p *KVStoreService) Watch(timeoutSecond int, keyChanged *string) error {
  2. id := fmt.Sprintf("watch-%s-%03d", time.Now(), rand.Int())
  3. ch := make(chan string, 10) // buffered
  4. p.mu.Lock()
  5. p.filter[id] = func(key string) { ch <- key }
  6. p.mu.Unlock()
  7. select {
  8. case <-time.After(time.Duration(timeoutSecond) * time.Second):
  9. return fmt.Errorf("timeout")
  10. case key := <-ch:
  11. *keyChanged = key
  12. return nil
  13. }
  14. return nil
  15. }

Watch方法的输入参数是超时的秒数。当有key变化时将key作为返回值返回。如果超过时间后依然没有key被修改,则返回超时的错误。Watch的实现中,用唯一的id表示每个Watch调用,然后根据id将自身对应的过滤器函数注册到p.filter列表。

KVStoreService服务的注册和启动过程我们不再赘述。下面我们看看如何从客户端使用Watch方法:

  1. func doClientWork(client *rpc.Client) {
  2. go func() {
  3. var keyChanged string
  4. err := client.Call("KVStoreService.Watch", 30, &keyChanged)
  5. if err != nil {
  6. log.Fatal(err)
  7. }
  8. fmt.Println("watch:", keyChanged)
  9. } ()
  10. err := client.Call(
  11. "KVStoreService.Set", [2]string{"abc", "abc-value"},
  12. new(struct{}),
  13. )
  14. if err != nil {
  15. log.Fatal(err)
  16. }
  17. time.Sleep(time.Second*3)
  18. }

首先启动一个独立的Goroutine监控key的变化。同步的watch调用会阻塞,直到有key发生变化或者超时。然后在通过Set方法修改KV值时,服务器会将变化的key通过Watch方法返回。这样我们就可以实现对某些状态的监控。

4.3.3 反向RPC

通常的RPC是基于C/S结构,RPC的服务端对应网络的服务器,RPC的客户端也对应网络客户端。但是对于一些特殊场景,比如在公司内网提供一个RPC服务,但是在外网无法链接到内网的服务器。这种时候我们可以参考类似反向代理的技术,首先从内网主动链接到外网的TCP服务器,然后基于TCP链接向外网提供RPC服务。

以下是启动反向RPC服务的代码:

  1. func main() {
  2. rpc.Register(new(HelloService))
  3. for {
  4. conn, _ := net.Dial("tcp", "localhost:1234")
  5. if conn == nil {
  6. time.Sleep(time.Second)
  7. continue
  8. }
  9. rpc.ServeConn(conn)
  10. conn.Close()
  11. }
  12. }

反向RPC的内网服务将不再主动提供TCP监听服务,而是首先主动链接到对方的TCP服务器。然后基于每个建立的TCP链接向对方提供RPC服务。

而RPC客户端则需要在一个公共的地址提供一个TCP服务,用于接受RPC服务器的链接请求:

  1. func main() {
  2. listener, err := net.Listen("tcp", ":1234")
  3. if err != nil {
  4. log.Fatal("ListenTCP error:", err)
  5. }
  6. clientChan := make(chan *rpc.Client)
  7. go func() {
  8. for {
  9. conn, err := listener.Accept()
  10. if err != nil {
  11. log.Fatal("Accept error:", err)
  12. }
  13. clientChan <- rpc.NewClient(conn)
  14. }
  15. }()
  16. doClientWork(clientChan)
  17. }

当每个链接建立后,基于网络链接构造RPC客户端对象并发送到clientChan管道。

客户端执行RPC调用的操作在doClientWork函数完成:

  1. func doClientWork(clientChan <-chan *rpc.Client) {
  2. client := <-clientChan
  3. defer client.Close()
  4. var reply string
  5. err = client.Call("HelloService.Hello", "hello", &reply)
  6. if err != nil {
  7. log.Fatal(err)
  8. }
  9. fmt.Println(reply)
  10. }

首先从管道去取一个RPC客户端对象,并且通过defer语句指定在函数退出前关闭客户端。然后是执行正常的RPC调用。

4.3.4 上下文信息

基于上下文我们可以针对不同客户端提供定制化的RPC服务。我们可以通过为每个链接提供独立的RPC服务来实现对上下文特性的支持。

首先改造HelloService,里面增加了对应链接的conn成员:

  1. type HelloService struct {
  2. conn net.Conn
  3. }

然后为每个链接启动独立的RPC服务:

  1. func main() {
  2. listener, err := net.Listen("tcp", ":1234")
  3. if err != nil {
  4. log.Fatal("ListenTCP error:", err)
  5. }
  6. for {
  7. conn, err := listener.Accept()
  8. if err != nil {
  9. log.Fatal("Accept error:", err)
  10. }
  11. go func() {
  12. defer conn.Close()
  13. p := rpc.NewServer()
  14. p.Register(&HelloService{conn: conn})
  15. p.ServeConn(conn)
  16. } ()
  17. }
  18. }

Hello方法中就可以根据conn成员识别不同链接的RPC调用:

  1. func (p *HelloService) Hello(request string, reply *string) error {
  2. *reply = "hello:" + request + ", from" + p.conn.RemoteAddr().String()
  3. return nil
  4. }

基于上下文信息,我们可以方便地为RPC服务增加简单的登陆状态的验证:

  1. type HelloService struct {
  2. conn net.Conn
  3. isLogin bool
  4. }
  5. func (p *HelloService) Login(request string, reply *string) error {
  6. if request != "user:password" {
  7. return fmt.Errorf("auth failed")
  8. }
  9. log.Println("login ok")
  10. p.isLogin = true
  11. return nil
  12. }
  13. func (p *HelloService) Hello(request string, reply *string) error {
  14. if !p.isLogin {
  15. return fmt.Errorf("please login")
  16. }
  17. *reply = "hello:" + request + ", from" + p.conn.RemoteAddr().String()
  18. return nil
  19. }

这样可以要求在客户端链接RPC服务时,首先要执行登陆操作,登陆成功后才能正常执行其他的服务。