手记

通过Nginx实现gRPC服务的负载均衡 | gRPC双向数据流的交互控制系列(3)

第一步:下载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 mainimport (    "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 服务端口 stringfunc 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数据流服务,至此大功告成!



作者:阿狸不歌
链接:https://www.jianshu.com/p/611d586f58cd


0人推荐
随时随地看视频
慕课网APP