手撸golang GO与微服务 net.rpc之2

缘起

最近阅读 [Go微服务实战] (刘金亮, 2021.1)
本系列笔记拟采用golang练习之
gitee: https://gitee.com/ioly/learning.gooop

net/rpc

微服务中的进程间通信概述
对于进程间通信的技术，开发者有多种选择。
可以选择基于同步通信的通信机制，比如HTTP RESTful；
也可以选择基于异步通信的方式，Go语言提供了标准的net/rpc包以支持异步。

远程过程调用协议（Remote Procedure Call Protocol, RPC）,
是一种通过网络从远程计算机程序上请求服务，
而不需要了解底层网络技术的协议。

目标(Day 2)

• Day 1的rpc测试是短连接，频繁dial和close非常影响吞吐，改为长连接版本看看

设计

• TimeConnection：封装rpc.dial以提供长连接的rpc.Client句柄。 rpc.Client.Call方法是并发安全的， 因此允许我们使用共享的长连接。
• ITimeClientV2: 时间客户端接口
• tTimeClientV2：时间客户端的实现， 持有长连接的rpc.Client句柄，进行远程GetTime调用

单元测试

• common.go，封装单元测试的通用代码

package net_rpc

import "testing"

func fnAssertTrue(t *testing.T, b bool, msg string) {
    if !b {
        t.Fatal(msg)
    }
}

type CallLog struct {
    done bool
    cost int64
}

• net_rpc_v2_test.go
• 使用TimeConnection提供的共享的长连接
• 并发100/300/500/1000/10000/50000次rpc调用
• 统计失败次数和平均耗时

package net_rpc

import (
    "learning/gooop/net_rpc"
    "sync"
    "testing"
    "time"
)


func Test_NetRPC_V2(t *testing.T) {
    server := new(net_rpc.TimeServer)
    err := server.Serve(3333)
    if err != nil {
        t.Fatal(err)
    }
    time.Sleep(100 * time.Millisecond)


    fnTestRpcCall := func(client net_rpc.ITimeClientV2, log *CallLog) {
        t0 := time.Now().UnixNano()
        err, ret := client.GetTime()
        log.cost = time.Now().UnixNano() - t0
        log.done = err == nil

        if log.done {
            fnAssertTrue(t, ret > 0, "expecting ret>0")
        }
    }

    fnTestConcurrency := func(threads int) {
        logs := make([]*CallLog, threads)
        for i, _ := range logs {
            logs[i] = new(CallLog)
        }

        var g sync.WaitGroup

        conn := new(net_rpc.TimeConnection)
        _ = conn.Connect("localhost:3333", func(client net_rpc.ITimeClientV2) error {
            for i, _ := range logs {
                n := i
                g.Add(1)
                go func() {
                    fnTestRpcCall(client, logs[n])
                    g.Done()
                }()
            }

            g.Wait()
            return nil
        })


        var failed, max, avg int64 = 0, 0, 0
        for _, it := range logs {
            if !it.done {
                failed++
            }

            if it.cost > max {
                max = it.cost
            }

            avg += it.cost
        }
        avg = avg / int64(threads)

        maxf := float64(max) / float64(time.Millisecond/time.Nanosecond)
        avgf := float64(avg) / float64(time.Millisecond/time.Nanosecond)
        t.Logf("threads=%d, failed=%d, max=%fms, avg=%fms", threads, failed, maxf, avgf)
    }

    fnTestConcurrency(100)
    fnTestConcurrency(300)
    fnTestConcurrency(500)
    fnTestConcurrency(1000)
    fnTestConcurrency(10000)
    fnTestConcurrency(50000)
}

测试输出

比Day 1的短连接模式快一个数量级

$ go test -v *.go -test.run Test_NetRPC_V2
=== RUN   Test_NetRPC_V2
2021/03/25 13:51:55 rpc.Register: method "Serve" has 2 input parameters; needs exactly three
    net_rpc_v2_test.go:71: threads=100, failed=0, max=1.715795ms, avg=1.179659ms
    net_rpc_v2_test.go:71: threads=300, failed=0, max=6.225059ms, avg=4.792163ms
    net_rpc_v2_test.go:71: threads=500, failed=0, max=10.110459ms, avg=5.502403ms
    net_rpc_v2_test.go:71: threads=1000, failed=0, max=17.945680ms, avg=8.392062ms
    net_rpc_v2_test.go:71: threads=10000, failed=0, max=153.765575ms, avg=85.053883ms
    net_rpc_v2_test.go:71: threads=50000, failed=0, max=709.802299ms, avg=299.928400ms
--- PASS: Test_NetRPC_V2 (1.02s)
PASS
ok      command-line-arguments  1.031s

TimeConnection.go

• 封装rpc.dial以提供长连接的rpc.Client句柄。
• rpc.Client.Call方法是并发安全的， 因此允许我们使用共享的长连接。

package net_rpc

import "net/rpc"

type TimeConnection int

func (me *TimeConnection) Connect(serverAddress string, action func(client ITimeClientV2) error) error {
    conn, err := rpc.Dial("tcp", serverAddress)
    if err != nil {
        return err
    }
    defer conn.Close()

    return action(newTimeClientV2(conn))
}

ITimeClientV2

时间客户端接口

package net_rpc

type ITimeClientV2 interface {
    GetTime() (error, int64)
}

tTimeClientV2.go

持有长连接的rpc.Client指针，进行远程GetTime调用

package net_rpc

import "net/rpc"

type tTimeClientV2 struct {
    client *rpc.Client
}

func newTimeClientV2(client *rpc.Client) ITimeClientV2 {
    return &tTimeClientV2{ client, }
}

func (me *tTimeClientV2) GetTime() (error, int64) {
    var t int64 = 0
    err := me.client.Call("TimeServer.GetTime", 1, &t)
    if err != nil {
        return err, 0
    }

    return nil, t
}

(end)