gRPC 到远程服务器的带宽较慢

学习知识要善于思考，思考，再思考！今天golang学习网小编就给大家带来《gRPC 到远程服务器的带宽较慢》，以下内容主要包含等知识点，如果你正在学习或准备学习Golang，就都不要错过本文啦~让我们一起来看看吧，能帮助到你就更好了！

我有一个 grpc 服务，可以将文件从本地计算机传输到远程服务器，并且我注意到一些严重的带宽问题。平均而言，在一个连接共享多个流（通常约为 8 个）的情况下，下载速度约为 1mb/s。

服务器使用tls进行加密，但这似乎不是瓶颈，因为关闭tls对性能的影响可以忽略不计。我还尝试使用 iperf3 直接测试客户端和服务器之间的带宽，结果为 10mb/s。

connecting to host <host>, port <port>
[  7] local 10.0.0.112 port 59651 connected to <ip> port <port>
[ id] interval           transfer     bitrate
[  7]   0.00-1.00   sec  1.28 mbytes  10.7 mbits/sec
[  7]   1.00-2.00   sec   894 kbytes  7.35 mbits/sec
[  7]   2.00-3.00   sec   999 kbytes  8.17 mbits/sec
[  7]   3.00-4.00   sec  1.19 mbytes  10.0 mbits/sec
[  7]   4.00-5.00   sec   753 kbytes  6.17 mbits/sec
[  7]   5.00-6.00   sec  1.16 mbytes  9.67 mbits/sec
[  7]   6.00-7.00   sec  1.00 mbytes  8.44 mbits/sec
[  7]   7.00-8.00   sec  1.26 mbytes  10.5 mbits/sec
[  7]   8.00-9.00   sec  1.22 mbytes  10.2 mbits/sec
[  7]   9.00-10.00  sec  1.15 mbytes  9.66 mbits/sec
- - - - - - - - - - - - - - - - - - - - - - - - -
[ id] interval           transfer     bitrate
[  7]   0.00-10.00  sec  10.8 mbytes  9.09 mbits/sec                  sender
[  7]   0.00-10.00  sec  10.7 mbytes  8.95 mbits/sec                  receiver

客户端上传带宽约10mb/s，服务器下载约50mb/s（通过speedtest-cli）

traceroute 也没有显示任何有趣的内容...

traceroute to mikemeredith.ddns.net (108.52.111.249), 64 hops max, 72 byte packets
 1  10.0.0.1 (10.0.0.1)  2.195 ms  5.388 ms  1.385 ms
 2  <ip>  (<ip>)  8.256 ms  145.115 ms  19.025 ms
 3  <ip2> (<ip2>)  9.951 ms  9.471 ms  141.929 ms
 4  <ip3> (<ip3>)  18.389 ms  9.684 ms  12.248 ms
 5  <ip4> (<ip4>)  143.880 ms  25.077 ms  10.606 ms
 6  ae-13-ar01.capitolhghts.md.bad.comcast.net (68.87.168.61)  142.567 ms  137.153 ms  20.790 ms
 7  be-33657-cr02.ashburn.va.ibone.comcast.net (68.86.90.57)  14.326 ms  144.076 ms  22.957 ms
 8  be-1102-cs01.ashburn.va.ibone.comcast.net (96.110.32.169)  13.881 ms  144.756 ms  23.981 ms
 9  be-2107-pe07.ashburn.va.ibone.comcast.net (96.110.32.186)  20.203 ms  94.433 ms  23.034 ms
10  n-a.gw12.iad8.alter.net (152.179.50.205)  20.254 ms  278.023 ms  31.660 ms
11  * * *
12  <ip12> (<ip13>)  66.277 ms  39.229 ms  34.543 ms
13  <ip13> (<ip14>)  48.849 ms  49.300 ms  49.546 ms

这是实际的代码

客户端连接：

creds, err := credentials.newclienttlsfromfile(cerloc, "")
if err != nil {
    fmt..printf("failed to get tls from file: %v\n", err)
    panic(err)
}
conn, err = grpc.dial(host+port, grpc.withtransportcredentials(creds))

客户端流：

ctx, cancel := context.withcancel(context.background())
defer cancel()
client := proto.client(conn)
stream, err := client.backupfiles(ctx, grpc.usecompressor(gzip.name))
// send on stream, max size of message is 2mb

服务器监听：

// Start serving on port
l, err := net.Listen("tcp", port)
if err != nil {
    fmt.Printf("error listening on port %v: %v\n", port, err)
    panic(err)
}

var s *grpc.Server
creds, err := credentials.NewServerTLSFromFile(
    certLoc,
    keyLoc,
)
if err != nil {
    fmt.Printf("error getting tls certs: %v\n", err)
    panic(err)
}
s = grpc.NewServer(grpc.Creds(creds))
proto.RegisterBackupServer(s, &server{})
err = s.Serve(l)


// Actual stream handling

// Get a pooled SharedBuffer for assembling the file
b := getBuffer()
defer putBuffer(b)
c := make(chan int, []50)
u, _ := user.Current()

uid := uuid.New()
fout, err := os.Create(filePath + uid.String())
if err != nil {
    fmt.Println("error creating staging file: ", err)
    panic(err)
}
var wg sync.WaitGroup

go assemble(b, fout, c, &wg)
for {
    in, err := stream.Recv()
    if err == io.EOF {
        break
    }
    if err != nil {
        fmt.Println("encountered an error receiving on stream: ", err)
        return err
    }


    bytesWritten = b.LockWrite(in.Payload) // This buffer is shared between the stream and the go routine

    c <- bytesWritten
}


close(c)
wg.Wait()

_ = fout.Close()

// This is a pre-declared workerpool that basically moves files around 
wp.Submit(func() {
    finalizeFile(fout.Name(), name, perms, "", checkSum, userID)
})

return stream.SendAndClose(&proto.Resp{
    Status:   true,
})

func assemble(b *buffer.SharedBuffer, fout *os.File, in chan int, wg *sync.WaitGroup) {
    wg.Add(1)
    defer wg.Done()

    buf := make([]byte, buffer.BUFFSIZE*2)

    for i := range in {
        if fout != nil {
            b.Lock()
            _, err := b.Read(buf[:i])
            b.Unlock()
            if err == io.EOF {
                continue
            }
            if err != nil {
                panic(err)
            }
            n, err := fout.Write(buf[:i])
            if err != nil {
                panic(err)
            }
            if n != i {
                fmt.Printf("failed to write all bytes to file: %v != %v", n, i)
                panic(err)
            }
        }
    }
}

似乎我可能遗漏了 grpc 内部工作原理的一些东西？

解决方案

如果我没猜错的话，您正在读取一个 goroutine 中的输入流并将字节分派到第二个 goroutine。为什么不让第二个 goroutine 完全处理流呢？通过这种方式，第一个 goroutine 可以自由地处理后续流（如果有）。

通常的模式是让一个 goroutine 监听传入的请求并生成新的 goroutine 来处理它们。至关重要的是，除了监听请求的 api 之外，主 goroutine 不调用任何阻塞 api。

例如：

for {
    newStream := ListenForStreams() //Block until next stream
    go consumeStream(newStream)
}

今天带大家了解了的相关知识，希望对你有所帮助；关于Golang的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~