前情提要
本系列的第一篇文章 通过一个例子介绍了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交互
网友评论