为什么是实践应用(1)?因为每一个软件都有很多事儿,稍后我们会拆分章节分到每一期去整理和讲解。

目前我们有一部分内部服务是通过 gRPC 构建的,这里主要讲下 gRPC 在我们这的一点实践,主要分下面几个主题,帮助大家了解我们的是如何在使用和部署 gRPC 服务的,帮助大家理一下思路。下面这个图,是当前我们 gRPC 服务相关的架构图,之后会针对每一块给大家做个介绍。本期意在讲述我们的架构部署和对应的服务有哪些服务构建,具体细节分期以后再讲。

gRPC 服务架构图

gRPC

RPC(Remote Procedure Call)一种远程调用协议,本身是一种语言无关的协议,简单说就是本地调用远程方法,而这种调用非常简单,对于本地试用者就像调用本地方法一样。关于 RPC 的服务框架有很多,这些框架使得使用 RPC 服务的人,更加专注业务,而不用更多的去关注通信、协议等实现,例如你调用了一个远程方法获取用户信息,在框架范围内只需要在本地执行 this->callRemoteUserInfoByUserId(123456) 就完成了用户信息的获取。 gRPC 也是一种 RPC 服务框架,类似的还有 Facebook 的 thrift,我们的服务扣费等功能,就是 thrift 实现的,还有频次控制服务也是 thrift 的。关于 gRPC 的文档说明和定义,都有了中文翻译版本这里不在复述。

protobuf

因为 RPC 服务是语言无关的,所以在不同语言上,就需要一个介质进行不同语言之间的数据转换,一般可以理解为不同语言间的序列化和反序列化,现在最常用的可能是 json 格式,例如 php 服务端接口生成的 json 为客户端 android 或者 iOS 提供服务。又如 RPC 服务 java 做服务端,php 做客户端,那么 php 和 java 之间就需要进行数据格式转换,Protocol Buffers 就是 Google 开源的一套数据转换格式(或者称之为协议),而且 Google 也提供了全套的工具,可以根据 protocol buffers 的数据 .proto 文件,生成对应的客户端和服务端代码,让不同编程语言之间通过该格式进行数据转换,而这种转换完全又工具生成,我们只关注服务及方法调用和自己的业务,让开发人员从代码、网络交互等工作中解脱出来。

既然是一种协议或者是数据结构,那就又定义这种结构的语法,这里有一个关于 .proto 语法的翻译的中文文档,(顺便说下这个博客的作者是位技术大牛,上面有很多非常不错的技术文章。),稍后的例子中,我们也会提供一个 .proto 的文件。

Linkerd

Linkerd 是一款开源网络代理,旨在作为服务网格进行部署,用于在应用程序内管理、控制和监视服务到服务通信的专用层,也是一套服务治理框架,他提供了性能看板等服务,通过性能看板我们可以直观的看到当前服务的状态。还有很多更强大的功能,包括负载均衡、服务发现等,也提供了多种部署方式,(之前技术部的微服务分享中有讲过,这里不在复述)。和其类似比较火的还有Istio

Consul

Consul 主要是提供服务发现、健康检查等服务的服务,何其相似的软件又大名鼎鼎的 zookeeperetcd 等。关于服务发现等软件解决的问题不用说太多,大家也都应该了解了。

示例

目前我们已经在使用一些 RPC 服务了,就像前面说的线上已经部署了一些 thrift 的服务,还有之前我们推荐服务的也是 gRPC 服务,之前我们一直是作为客户端使用 gRPC 服务,这次的例子主要是客户端和服务端都涉及。当前线上的服务可以通过 host:9990 来查看 dashboard。

下面我们围绕一个 ApplyService 服务来讲解,看看完成一个服务需要我们做哪些事情。

完成一个服务需要我们做哪些事情?

基础环境安装

客户端

  • php 的 grpc 扩展
  • php 的 composer 扩展包 grpc/grpc google/protobuf

服务端

  • go
  • grpc-go

proto 工具

  • protoc 代码生成工具和对应的语言插件 grpc_php_plugin protoc-gen-go 等

定义 proto 文件

$ cat services/apply/apply.proto

syntax = "proto3";

// package 名称,因为不同的语言在 package 管理上方式不同,
// 所以 proto 提供了 option 关键字,来帮助不同语言之间让 package 更加友好
// 因为我们是把这部分服务单独放在项目里的,通过客户端通过 composer 统一管理,
// 所以用了单独的目录和独立的名称空间 pluto,php 生成的代码名称空间就是
// namespace Pluto\Services\Apply;
// 在 composer.json 内加入 autoload 模块
// "autoload" : {
//    "psr-4": {
//        "Pluto\\": "grpc/Pluto/",
//    }
// }
package pluto.services.apply;

// 为了上面所说的生成 golang 友好的 package 而设定的
option go_package = "apply";

service ApplyService {

    // 一个 rpc 调用
    rpc FetchApplyByUid(ApplyRequest) returns (ApplyResponse) {};
}

// 请求
message ApplyRequest {
    string uid = 1;
}

// 响应
message ApplyResponse {
   string result = 1;
}

除了正常的业务定义以外,我们跟多方合作时,为了语言上的友好,尽量通过 option 关键字,来定义语言友好的包名,让代码组织起来不混乱。

客户端代码生成

protoc --php_out=grpc/ --grpc_out=grpc/ --plugin=protoc-gen-grpc=/usr/local/bin/grpc_php_plugin resource/apply.proto

protoc-gen-grpc 这个插件完成了 Client 类的生成,如果不安装该插件也可以生成代码,但是 Client 里面的逻辑要自己去加入。除了生成的基础类以外,我们看下 Client 这个文件

<?php
// GENERATED CODE -- DO NOT EDIT!

namespace Pluto\Services\Apply;

/**
 */
class ApplyServiceClient extends \Grpc\BaseStub {

    /**
     * @param string $hostname hostname
     * @param array $opts channel options
     * @param \Grpc\Channel $channel (optional) re-use channel object
     */
    public function __construct($hostname, $opts, $channel = null) {
        parent::__construct($hostname, $opts, $channel);
    }

    /**
     * @param \Pluto\Services\Apply\ApplyRequest $argument input argument
     * @param array $metadata metadata
     * @param array $options call options
     */
    public function FetchApplyByUid(\Pluto\Services\Apply\ApplyRequest $argument,
      $metadata = [], $options = []) {

        // 注意这一段代码
        // pluto.services.apply.ApplyService 这个就是服务名称,
        // 我们后续会使用到这个名字
        return $this->_simpleRequest('/pluto.services.apply.ApplyService/FetchApplyByUid',
        $argument,
        ['\Pluto\Services\Apply\ApplyResponse', 'decode'],
        $metadata, $options);
    }

}

PHP 客户端调用代码

// 这里需要注意 php 的 grpc 扩展的版本,及时升级最新版,
// 笔者开始时测试 10w 次请求发现客户端内存飙升,查了 github issue 发现是因为有内存泄漏
// 官方在新版本修复了
private function callGrpc($id, $output)
{
    // 请求对象初始化
    $request = new ApplyRequest();
    $request->setUid($id);

    // 初始化 Client
    // 此处配置的是 linkerd 的 4141 代理端口,我们服务用的是 9527 端口
    // 稍后会说明
    $client = new ApplyServiceClient("127.0.0.1:4141", [
        'credentials' =>ChannelCredentials::createInsecure(),
        'timeout' => 1000000,
    ]);

    // 调用服务
    list($reply, $status) = $client->FetchApplyByUid($request)->wait();
    var_dump($reply->getResult());
    $output->writeln($reply->getResult());
}

服务单 golang 代码生成

根据自己组织的目录进行代码生成,根据我们的定义,会生成一个 apply.pb.go 文件

protoc --go_out=plugins=grpc:. *.proto

先整体看下我们 golang 服务端的代码结构

.
├── Makefile
├── README.md
├── bin // 编译后的二进制文件目录
│   └── pluto
├── config // 配置文件
│   └── config.yaml
├── glide.lock
├── glide.yaml
├── logs
├── main.go // 主程序
├── services
│   ├── apply
│   └── apply.go // ApplyService 的业务代码
├── util // 工具等目录
│   ├── init.go
│   └── storage.go
└── vendor // 依赖
    ├── github.com
    ├── golang.org
    ├── google.golang.org
    └── gopkg.in

$ cat services/apply/apply.pb.go 注意这部分代码 ServiceName 的注释部分

// Code generated by protoc-gen-go. DO NOT EDIT.
// source: apply.proto

/*
Package apply is a generated protocol buffer package.

It is generated from these files:
    apply.proto

It has these top-level messages:
    ApplyRequest
    ApplyResponse
*/
package apply

import proto "github.com/golang/protobuf/proto"
import fmt "fmt"
import math "math"

import (
    context "golang.org/x/net/context"
    grpc "google.golang.org/grpc"
)

// Reference imports to suppress errors if they are not otherwise used.
var _ = proto.Marshal
var _ = fmt.Errorf
var _ = math.Inf

// This is a compile-time assertion to ensure that this generated file
// is compatible with the proto package it is being compiled against.
// A compilation error at this line likely means your copy of the
// proto package needs to be updated.
const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package

type ApplyRequest struct {
    Uid string `protobuf:"bytes,1,opt,name=uid" json:"uid,omitempty"`
}

func (m *ApplyRequest) Reset()                    { *m = ApplyRequest{} }
func (m *ApplyRequest) String() string            { return proto.CompactTextString(m) }
func (*ApplyRequest) ProtoMessage()               {}
func (*ApplyRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0} }

func (m *ApplyRequest) GetUid() string {
    if m != nil {
        return m.Uid
    }
    return ""
}

type ApplyResponse struct {
    Result string `protobuf:"bytes,1,opt,name=result" json:"result,omitempty"`
}

func (m *ApplyResponse) Reset()                    { *m = ApplyResponse{} }
func (m *ApplyResponse) String() string            { return proto.CompactTextString(m) }
func (*ApplyResponse) ProtoMessage()               {}
func (*ApplyResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1} }

func (m *ApplyResponse) GetResult() string {
    if m != nil {
        return m.Result
    }
    return ""
}

func init() {
    proto.RegisterType((*ApplyRequest)(nil), "pluto.services.apply.ApplyRequest")
    proto.RegisterType((*ApplyResponse)(nil), "pluto.services.apply.ApplyResponse")
}

// Reference imports to suppress errors if they are not otherwise used.
var _ context.Context
var _ grpc.ClientConn

// This is a compile-time assertion to ensure that this generated file
// is compatible with the grpc package it is being compiled against.
const _ = grpc.SupportPackageIsVersion4

// Client API for ApplyService service

type ApplyServiceClient interface {
    FetchApplyByUid(ctx context.Context, in *ApplyRequest, opts ...grpc.CallOption) (*ApplyResponse, error)
}

type applyServiceClient struct {
    cc *grpc.ClientConn
}

func NewApplyServiceClient(cc *grpc.ClientConn) ApplyServiceClient {
    return &applyServiceClient{cc}
}

func (c *applyServiceClient) FetchApplyByUid(ctx context.Context, in *ApplyRequest, opts ...grpc.CallOption) (*ApplyResponse, error) {
    out := new(ApplyResponse)
    err := grpc.Invoke(ctx, "/pluto.services.apply.ApplyService/FetchApplyByUid", in, out, c.cc, opts...)
    if err != nil {
        return nil, err
    }
    return out, nil
}

// Server API for ApplyService service

type ApplyServiceServer interface {
    FetchApplyByUid(context.Context, *ApplyRequest) (*ApplyResponse, error)
}

func RegisterApplyServiceServer(s *grpc.Server, srv ApplyServiceServer) {
    s.RegisterService(&_ApplyService_serviceDesc, srv)
}

func _ApplyService_FetchApplyByUid_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
    in := new(ApplyRequest)
    if err := dec(in); err != nil {
        return nil, err
    }
    if interceptor == nil {
        return srv.(ApplyServiceServer).FetchApplyByUid(ctx, in)
    }
    info := &grpc.UnaryServerInfo{
        Server:     srv,
        FullMethod: "/pluto.services.apply.ApplyService/FetchApplyByUid",
    }
    handler := func(ctx context.Context, req interface{}) (interface{}, error) {
        return srv.(ApplyServiceServer).FetchApplyByUid(ctx, req.(*ApplyRequest))
    }
    return interceptor(ctx, in, info, handler)
}

var _ApplyService_serviceDesc = grpc.ServiceDesc{
    // 注意这部分代,ServiceName
    // 与之前 PHP 客户端里的生成代码部分 pluto.services.apply.ApplyService
    // 代表了服务的名称,这个名字稍后会用于服务发现服务的注册
    ServiceName: "pluto.services.apply.ApplyService",
    HandlerType: (*ApplyServiceServer)(nil),
    Methods: []grpc.MethodDesc{
        {
            MethodName: "FetchApplyByUid",
            Handler:    _ApplyService_FetchApplyByUid_Handler,
        },
    },
    Streams:  []grpc.StreamDesc{},
    Metadata: "apply.proto",
}

func init() { proto.RegisterFile("apply.proto", fileDescriptor0) }

var fileDescriptor0 = []byte{
    // 163 bytes of a gzipped FileDescriptorProto
    0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0x4e, 0x2c, 0x28, 0xc8,
    0xa9, 0xd4, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x12, 0x29, 0xc8, 0x29, 0x2d, 0xc9, 0xd7, 0x2b,
    0x4e, 0x2d, 0x2a, 0xcb, 0x4c, 0x4e, 0x2d, 0xd6, 0x03, 0xcb, 0x29, 0x29, 0x70, 0xf1, 0x38, 0x82,
    0x18, 0x41, 0xa9, 0x85, 0xa5, 0xa9, 0xc5, 0x25, 0x42, 0x02, 0x5c, 0xcc, 0xa5, 0x99, 0x29, 0x12,
    0x8c, 0x0a, 0x8c, 0x1a, 0x9c, 0x41, 0x20, 0xa6, 0x92, 0x3a, 0x17, 0x2f, 0x54, 0x45, 0x71, 0x41,
    0x7e, 0x5e, 0x71, 0xaa, 0x90, 0x18, 0x17, 0x5b, 0x51, 0x6a, 0x71, 0x69, 0x4e, 0x09, 0x54, 0x15,
    0x94, 0x67, 0x94, 0x03, 0x35, 0x2a, 0x18, 0x62, 0x83, 0x50, 0x0c, 0x17, 0xbf, 0x5b, 0x6a, 0x49,
    0x72, 0x06, 0x58, 0xd0, 0xa9, 0x32, 0x34, 0x33, 0x45, 0x48, 0x49, 0x0f, 0x9b, 0x23, 0xf4, 0x90,
    0x5d, 0x20, 0xa5, 0x8c, 0x57, 0x0d, 0xc4, 0x0d, 0x4a, 0x0c, 0x4e, 0xec, 0x51, 0xac, 0x60, 0x89,
    0x24, 0x36, 0xb0, 0xf7, 0x8c, 0x01, 0x01, 0x00, 0x00, 0xff, 0xff, 0x1d, 0x1a, 0x00, 0x10, 0xed,
    0x00, 0x00, 0x00,
}

服务端代码

// 指定默认端口,可以通过启动命令行修改
// 版本为了编译后查看和服务替换时进行检查
const (
    appListen  = ":9527"
    appVersion = "1.0.0"
    appName    = "Pluto"
)

// 服务可以通过指定配置文件来启动,基础的配置目前还是依赖配置文件
// 服务启动可以指定两个参数:
// -listen=host:port 指定监听端口
// -config=/your/config/path 指定配置文件
// -version 并不会启动服务,只是会输出一些基础数据,用于查看版本等。
var GlobalConfPath string
var GlobalListen string
var GlobalVersion bool

func init() {
    flag.StringVar(&GlobalConfPath, "config", "./config/config.yaml", "Config.yaml")
    flag.StringVar(&GlobalListen, "listen", appListen, "Host:port")
    flag.BoolVar(&GlobalVersion, "version", false, "Version")
    flag.Parse()

    util.InitConfigure(GlobalConfPath)
}

// 退出系统
func exitApplication() {
    log.Println("Exit success...")
    os.Exit(0)
}

func main() {
    log.Println("Starting...")

    // 如果参数是 -version 则只是数据配置
    if GlobalVersion {
        fmt.Println(appName, ":", appVersion)
        fmt.Println("Listen:", GlobalListen)
        fmt.Println("Config:", GlobalConfPath)
        os.Exit(0)
    }

    // 创建一个信号监听,如果向该服务发送一些信号,对应作出处理,
    // 这里主要是退出服务
    notifySignal := make(chan os.Signal)
    signal.Notify(notifySignal, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGUSR1, syscall.SIGUSR2)
    go func() {
        for signalResult := range notifySignal {
            switch signalResult {
            case syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGUSR1, syscall.SIGUSR2:
                exitApplication()
            }
        }
    }()

    log.Println("Listening....")

    listen, err := net.Listen("tcp", GlobalListen)
    if err != nil {
        log.Fatalf("Failed to listen: %v", err)
    }

    // grpc 注册相应的 rpc 服务
    server := grpc.NewServer()
    pb.RegisterApplyServiceServer(server, &service.ApplyService{})
    reflection.Register(server)

    log.Println("Starting success...")
    //  开启服务
    if err := server.Serve(listen); err != nil {
        log.Fatalf("Failed to serve: %v", err)
    }
}

至此,客户端和服务端的代码都完成了,我们部署了 linkerd 和 consul 服务,所以需要我们注册服务到 consul 上,之后我们就可以通过 127.0.0.1:4141 端口来调用了,当然直接调用服务的 9527 端口也是没问题的,但是我们是集群服务,避免单点,所以需要把服务注册到 consul 上。注册服务我们可以通过代码也可以通过我们的界面管理,因为我们有统一的 webUI 管理界面,所以我们直接通过 webUI 的界面来注册。注册服务的时候有一个非常重要的字段 ServiceName,就是我们前文两次提到的那个名字,只有通过这个名字,服务端和客户端才能通过 linkerd 的代理服务建立起链接。而这一切 linkerd 和 consul 都为我们做好了,我们只需要将服务注册上去即可。

服务注册

PS: 由于咱们墙大,所以 golang 的很多包无法下载,没有代理的话,可以通过 glide mirror 功能设置镜像,大部分 google 域名下的包在 github 都有镜像,可以顺利下载。这个功能太好用了,再补充一个 glidle.yaml 的配置文件内容吧。

import:
- package: github.com/golang/protobuf
  version: ^1.0.0
  subpackages:
  - proto
- package: golang.org/x/net
  subpackages:
  - context
- package: golang.org/x/text
  subpackages:
  - unicode
  - secure
- package: google.golang.org/grpc
  subpackages:
  - reflection
- package: google.golang.org/genproto
  subpackages:
  - rpc/status
  - googleapis
- package: google.golang.org/appengine
  subpackages:
  - cloudsql
- package: github.com/olivere/elastic
  version: ^2.0.60
- package: github.com/jmoiron/sqlx
- package: github.com/bradfitz/gomemcache
  subpackages:
  - memcache
- package: golang.org/x/lint
- package: github.com/jinzhu/configor
- package: github.com/BurntSushi/toml
  version: ^0.3.0

下面是 .glidle/mirrors.yaml 配置

repos:
- original: https://golang.org/x/crypto
  repo: https://github.com/golang/crypto
- original: https://golang.org/x/lint
  repo: https://github.com/golang/lint
- original: https://golang.org/x/net
  repo: https://github.com/golang/net
- original: https://golang.org/x/sys
  repo: https://github.com/golang/sys
- original: https://golang.org/x/text
  repo: https://github.com/golang/text
- original: https://google.golang.org/appengine
  repo: https://github.com/golang/appengine
- original: https://google.golang.org/genproto
  repo: https://github.com/google/go-genproto
- original: https://google.golang.org/grpc
  repo: https://github.com/grpc/grpc-go

EOT;