前情提要

本系列的第一篇文章 通过一个例子介绍了go语言实现gRPC双向数据流的交互控制,第二篇文章介绍了如何通过Websocket与gRPC交互。通过这两篇文章,我们可以一窥gRPC双向数据流的开发方式,但是在生产环境当中一台服务器(一个服务端程序)是不够的,我们往往会面临各种复杂情况:访问量上来了一台服务器不够怎么办?服务器挂了怎么办?有实战经验的读者肯定知道答案:上负载均衡(Load Balancing)啊!

gRPC服务如何做负载均衡?

gRPC官方博客上有一篇文章《gRPC Load Balancing》(https://grpc.io/blog/loadbalancing),详细介绍了几种方案,并分析了几种方案各自的优劣。并附了一张解决方案表: gRPC负载均衡解决方案表

在gRPC的Github上还有一篇文章叫《Load Balancing in gRPC》(https://github.com/grpc/grpc/blob/master/doc/load-balancing.md),如果英文看着费劲可以看一篇中文的《gRPC服务发现&负载均衡》(https://segmentfault.com/a/1190000008672912)。


测试Nginx对gRPC服务的支持

因为上面几篇文章介绍的很详细了,所以本文不再展开讨论。我们可以注意到上表中被红框圈起来的部分写着“Nginx coming soon”,现在这个Nginx的解决方案已经来了——2018年3月17日,Nginx官方宣布nginx 1.13.10支持gRPC (https://www.nginx.com/blog/nginx-1-13-10-grpc/)

第一步:下载nginx最新的stable版(本文发稿时是1.14.0,如果会用docker的也可以下载其alpine版本)。 第二步:配置nginx的config文件如下

server {
# ⚠️ nginx的监听端口按你的实际情况设置
listen  80     http2;
access_log /var/log/nginx/access.log main;
location / {
        # ⚠️ 把下面的 grpc://127.0.0.1:3000换成你自己的grpc服务器地址
        grpc_pass grpc://127.0.0.1:3000;
    }
}

第三步:把go语言实现gRPC双向数据流的交互控制 一文中的client.go 中的服务端地址改为nginx服务的地址(比如:127.0.0.1:80)

第四步: (1)运行server.go (2)运行nginx服务 (3)运行client.go

如果没什么意外,gRPC客户端发出的消息可以通过nginx后被gRPC服务端收到。 nginx日志

我们可以通过nginx日志观察到相应的信息。

一个小坑

上述连接虽然已经实现,但是如果我们的客户端有连续一分钟没有输入信息,会出现接收信息出错的情况。 连接被nginx断开

这种情形在没有使用nginx的时候不会出现,由于以前使用nginx给websocket做反向代理时也出现过类似情况,故而推断是nginx对超过一段时间的连接进行了断开。

添加心跳

解决上述问题可以采取的一个方法是增加心跳(如果您发现了什么别的好办法可以解决这个问题,比如在nginx里配置一些参数,请留言告诉我😄

client.go

添加一段隔40秒发送心跳的代码

package main
import (
    "bufio"
    "context"
    "flag"
    "io"
    "log"
    "os"
    "time"
    "google.golang.org/grpc"
    proto "chat" // 根据proto文件自动生成的代码
)
var 服务器地址 string
func init() {
    flag.StringVar(&服务器地址, "server", "127.0.0.1:80", "服务器地址")
}
func main() {
    // 创建连接
    conn, err := grpc.Dial(服务器地址, grpc.WithInsecure())
    if err != nil {
        log.Printf("连接失败: [%v]\n", err)
        return
    }
    defer conn.Close()
    client := proto.NewChatClient(conn)
    // 声明 context
    ctx := context.Background()
    // 创建双向数据流
    stream, err := client.BidStream(ctx)
    if err != nil {
        log.Printf("创建数据流失败: [%v]\n", err)
        return
    }
    // 启动一个 goroutine 接收命令行输入的指令
    go func() {
        log.Println("请输入消息...")
        输入 := bufio.NewReader(os.Stdin)
        for {
            // 获取 命令行输入的字符串, 以回车 \n 作为结束标志
            命令行输入的字符串, _ := 输入.ReadString('\n')
            // 向服务端发送 指令
            if err := stream.Send(&proto.Request{Input: 命令行输入的字符串}); err != nil {
                return
            }
        }
    }()
    //⚠️ 新添加的部分: 启动一个 goroutine 每隔40秒发送心跳包
    go func() {
        for {
            // 每隔 40 秒发送一次
            time.Sleep(40 * time.Second)
            log.Println("发送心跳包")
            // 心跳字符用"\n"
            if err := stream.Send(&proto.Request{Input: "\n"}); err != nil {
                return
            }
        }
    }()
    for {
        // 接收从 服务端返回的数据流
        响应, err := stream.Recv()
        if err == io.EOF {
            log.Println("⚠️ 收到服务端的结束信号")
            break
        }
        if err != nil {
            // TODO: 处理接收错误
            log.Println("接收数据出错:", err)
            break
        }
        log.Printf("[客户端收到]: %s", 响应.Output)
    }
}

server.go

添加一段检测心跳的代码

package main
import (
    "flag"
    "io"
    "log"
    "net"
    "strconv"
    "google.golang.org/grpc"
    proto "chat" // 根据proto文件自动生成的代码
)
// Streamer 服务端
type Streamer struct{}
// BidStream 实现了 ChatServer 接口中定义的 BidStream 方法
func (s *Streamer) BidStream(stream proto.Chat_BidStreamServer) error {
    ctx := stream.Context()
    for {
        select {
        case <-ctx.Done():
            log.Println("收到客户端通过context发出的终止信号")
            return ctx.Err()
        default:
            // 接收从客户端发来的消息
            输入, err := stream.Recv()
            if err == io.EOF {
                log.Println("客户端发送的数据流结束")
                return nil
            }
            if err != nil {
                log.Println("接收数据出错:", err)
                return err
            }
            // 如果接收正常,则根据接收到的 字符串 执行相应的指令
            switch 输入.Input {
            case "结束对话\n", "结束对话":
                log.Println("收到'结束对话'指令")
                if err := stream.Send(&proto.Response{Output: "收到结束指令"}); err != nil {
                    return err
                }
                // 收到结束指令时,通过 return nil 终止双向数据流
                return nil
            case "返回数据流\n", "返回数据流":
                log.Println("收到'返回数据流'指令")
                // 收到 收到'返回数据流'指令, 连续返回 10 条数据
                for i := 0; i < 10; i++ {
                    if err := stream.Send(&proto.Response{Output: "数据流 #" + strconv.Itoa(i)}); err != nil     {
                        return err
                    }
                }
            // ⚠️ 拦截心跳字符"\n"
            case "\n":
                log.Println("收到心跳包")
                // 只接收心跳不回发数据也可以
            default:
                // 缺省情况下, 返回 '服务端返回: ' + 输入信息
                log.Printf("[收到消息]: %s", 输入.Input)
                if err := stream.Send(&proto.Response{Output: "服务端返回: " + 输入.Input}); err != nil {
                    return err
                }
            }
        }
    }
}
var 服务端口 string
func init() {
    flag.StringVar(&服务端口, "port", "3000", "服务端口")
}
func main() {
    log.Println("启动服务端...")
    server := grpc.NewServer()
    // 注册 ChatServer
    proto.RegisterChatServer(server, &Streamer{})
    address, err := net.Listen("tcp", ":"+服务端口)
    if err != nil {
        panic(err)
    }
    if err := server.Serve(address); err != nil {
        panic(err)
    }
}

添加完成后再度测试,连接不会再被nginx打断。


Nginx实现服务端负载均衡的配置文件

心跳的坑趟过去之后,剩下的其实就简单了,我们修改nginx的配置文件:

upstream backend {
    # ⚠️ 把下面的服务端地址和端口改成你自己的
    server 127.0.0.1:3000;
    server 127.0.0.1:3001;
} 
server {
    listen  80     http2;
    access_log /var/log/nginx/access.log main;
    location / {
        grpc_pass grpc://backend;
    }
}

按如下顺序启动 (1)运行多个 server.go ,按照nginx配置文件输入端口参数(如 server.go -port 3001)

(2)运行nginx服务

(3)运行多个**client.go, (也可以运行websocket的那个程序,记得把心跳代码加上**,多开几个浏览器窗口)

我们可以观察到开启的多个server都在进行gRPC数据流服务,至此大功告成🏆!


总结

gRPC服务端的负载均衡有很多种方案,也各有优劣,但是用Nginx似乎是最简单的一种。总之,我们还得根据具体的业务场景来选择具体的实现方案。


gRPC双向数据流系列

(之一): gRPC双向数据流的交互控制

(之二): 通过Websocket与gRPC交互