- 强大的工具支持，尽可能少的代码编写
- 极简的接口
- 完全兼容 net/http
- 支持中间件，方便扩展
- 高性能
- 面向故障编程，弹性设计
- 内建服务发现、负载均衡
- 内建限流、熔断、降载，且自动触发，自动恢复
- API 参数自动校验
- 超时级联控制
- 自动缓存控制
- 链路跟踪、统计报警等
- 高并发支撑，稳定保障了晓黑板疫情期间每天的流量洪峰
docker run -d --name Etcd-server --network app-tier --publish 2379:2379 --publish 2380:2380 --env ALLOW_NONE_AUTHENTICATION=yes --env ETCD_ADVERTISE_CLIENT_URLS=http://etcd-server:2379 bitnami/etcd:latest
//查看当前安装版本校验是否安装启动成功
etcdctl --version
//设置API version的版本设为3(官方建议,并且3和2命令和功能方面有不少的差别)
set ETCDCTL_API=3
//通过put存储(下方表示存储了一个key为hello,值为world,存储成功显示ok)
etcdctl put hello world
//get取值
etcdctl get hello
- 安装配置golang开发环境
- 安装goctl
goctl api new 服务名称
goctl api go -api user.api -dir . -style go_zero
- 创建一个user.proto文件,基于这个文件生成rpc服务
- 会生成一个名为User的rpcService
- 这个rpc服务中存在一个rpc接口Auth()
- 并且定义了Auth()接口的入参结构UserAuthReq, 反参结构UserAuthResp
syntax = "proto3";
package template;
//指定文件生成到哪个目录
option go_package = "./user";
//1.编写rpc服务接口,定义了一个名为User的rpcService
service User{
//内部存在一个rpc接口,接口名为Auth()
//入参类型为: UserAuthReq
//反参类型为: UserAuthResp
rpc Auth(UserAuthReq) returns (UserAuthResp);
}
message UserAuthReq{
string token = 1;
}
message UserAuthResp{
string userName = 1;
string password = 2;
map<string, string> extend = 3;
}
$ goctl env check -i -f
goctl rpc protoc 文件名.proto --go_out=./types --go-grpc_out=./types --zrpc_out=. --style go_zero
- “xxx.pb.go”: 内部是编写这个rpc服务时自定义的结构体相关处理
- “xxx_grpc.pb.go”: 内部是启动这个rpc服务端,启动访问这个rpc客户端的相关代码
- 其中user.proto是用来生成rpc服务的IDL文件
- etc文件夹下的user.yaml是当前rpc服务运行时读取的配置文件(需要在内部配置etcd连接…)
- internal/config/config.go是用来解析配置文件的结构体
- internal/logic/auth_logic.go内部是对应Auth()这个接口的业务处理方法(业务方法内部需要自己编写)
- internal/svc/service_context.go内部是当前rpc服务运行时的ServiceContext上下文
- internal/server/user_server.go内部是用来配置启动当前rpc服务端相关的代码,用来启动当前rpc服务使用的
- xxxxClient/xxx.go(当前对应userClient/user.go)内部是用来配置访问当前这个rpc服务的客户端相关代码
- 会针对业务接口生成一个"internal/logic/xxx_logic.go"文件,
- 该文件中定义了业务结构体,该结构体上实现了业务方法,但是方法内部为空具体的业务逻辑需要自己实现,
- 并且提供了初始化这个结构体的函数
- 例如当前创建的rpc服务中存在一个Auth()接口,生成了auth_logic.go文件,查看这个文件
import (
"context"
"errors"
"github.com/zeromicro/go-zero/core/logx"
"go_cloud_demo/rpc/internal/svc"
"go_cloud_demo/rpc/types/user"
)
// 业务结构体
type AuthLogic struct {
ctx context.Context
svcCtx *svc.ServiceContext
logx.Logger
}
// 初始化结构体函数
func NewAuthLogic(ctx context.Context, svcCtx *svc.ServiceContext) *AuthLogic {
return &AuthLogic{
ctx: ctx,
svcCtx: svcCtx,
Logger: logx.WithContext(ctx),
}
}
// 业务方法
func (l *AuthLogic) Auth(in *user.UserAuthReq) (*user.UserAuthResp, error) {
if len(in.Token) == 0 {
return nil, errors.New("为获取到token")
}
return &user.UserAuthResp{
UserName: "admin",
Password: "12345",
}, nil
}
- 针对当前服务创建了一个结构体
- 该结构体上实现了当前rpc服务对外的所有接口
- 每个接口内部会调用"/logic"中对应的初始化函数,初始化实际的业务结构体,通过这个结构体执行实际的业务接口
import (
"context"
"go_cloud_demo/rpc/internal/logic"
"go_cloud_demo/rpc/internal/svc"
"go_cloud_demo/rpc/types/user"
)
// 服务结构体(一个服务中只有一个)
type UserServer struct {
svcCtx *svc.ServiceContext
user.UnimplementedUserServer
}
// 初始化服务结构体
func NewUserServer(svcCtx *svc.ServiceContext) *UserServer {
return &UserServer{
svcCtx: svcCtx,
}
}
// 实现Auth()业务方法
func (s *UserServer) Auth(ctx context.Context, in *user.UserAuthReq) (*user.UserAuthResp, error) {
//内部会调用初始化logic下实现了实际业务方法结构体函数,
//例如当前调用logic下NewAuthLogic(),
//初始化获取实现了Auth()业务接口的AuthLogic结构体
l := logic.NewAuthLogic(ctx, s.svcCtx)
//调用实际的业务接口
return l.Auth(in)
}
// 实现Test业务方法
func (s *UserServer) Test(ctx context.Context, in *user.UserAuthReq) (*user.UserAuthResp, error) {
l := logic.NewTestLogic(ctx, s.svcCtx)
return l.Test(in)
}
- 读取读取配置文件,将配置文件解析到Config结构体
- 调用NewServiceContext()函数,封装服务执行上下文
- 执行通过proto生成的"xxx_grpc_pb.go"文件中的RegisterXXXXServer()函数,进行服务注册
- 执行zrpc下的MustNewServer() 创建rpc服务
package main
import (
"flag"
"fmt"
"go_cloud_demo/rpc/internal/config"
"go_cloud_demo/rpc/internal/server"
"go_cloud_demo/rpc/internal/svc"
"go_cloud_demo/rpc/types/user"
"github.com/zeromicro/go-zero/core/conf"
"github.com/zeromicro/go-zero/core/service"
"github.com/zeromicro/go-zero/zrpc"
"google.golang.org/grpc"
"google.golang.org/grpc/reflection"
)
// 命令行参数读取配置文件所在路径
var configFile = flag.String("f", "rpc/etc/user.yaml", "the config file")
func main() {
flag.Parse()
//1.读取配置文件解析到Config结构体上
var c config.Config
conf.MustLoad(*configFile, &c)
//2.创房服务运行上下文
ctx := svc.NewServiceContext(c)
//3.将服务注册到rpc服务器,并且监听指定端口启动服务
//参数一"c.RpcServerConf":保存了当前rpc服务配置信息
//参数二"func(grpcServer *grpc.Server)"一个函数,当执行该函数时
//会调用通过proto生成的RegisterXXXServer(),将当前rpc服务实现注册到rpc服务器
s := zrpc.MustNewServer(c.RpcServerConf,
func(grpcServer *grpc.Server) {
user.RegisterUserServer(grpcServer, server.NewUserServer(ctx))
if c.Mode == service.DevMode || c.Mode == service.TestMode {
reflection.Register(grpcServer)
}
})
defer s.Stop()
fmt.Printf("Starting rpc server at %s...\n", c.ListenOn)
s.Start()
}
Name: user.rpc
ListenOn: 0.0.0.0:8080 #当前rpc服务访问地址
Etcd:
Hosts:
- 127.0.0.1:2379
Key: user.rpc
package config
import (
"github.com/zeromicro/go-zero/rest"
"github.com/zeromicro/go-zero/zrpc"
)
type Config struct {
rest.RestConf
zrpc.RpcClientConf
}
package main
import (
"context"
"github.com/zeromicro/go-zero/core/discov"
"github.com/zeromicro/go-zero/zrpc"
"go_cloud_demo/rpc/types/user"
"log"
)
func main2() {
//1.创建zrpc客户端
client := zrpc.MustNewClient(zrpc.RpcClientConf{
Etcd: discov.EtcdConf{
//etcd连接地址
Hosts: []string{"127.0.0.1:2379"},
//目标服务key
Key: "user.rpc",
},
})
//2.获取连接
conn := client.Conn()
//3.使用proto生成的模板文件,创建服务client客户端
UserClient := user.NewUserClient(conn)
//4.调用指定接口接收响应
resp, err := UserClient.Auth(context.Background(), &user.UserAuthReq{Token: "数据"})
if err != nil {
log.Fatal(err)
}
log.Println(resp.Password)
}
- 配置etcd地址,在etcd上获取rpc目标服务访问地址
- 解析配置文件的Config上添加保存rpc信息的属性
- 服务运行的ServiceContext上下文结构体上添加请求rpc服务所需要的上线文属性
- 查看上方根据go-zero针对rpc客户端生成的"/xxxclient/xxx.go"文件中的代码,编写rpc客户端
- 当前服务的所有请求都需要先执行一下目标服务的Auth()接口,将访问rpc服务的Auth()抽取为中间件
- 启动服务,访问接口进行测试
Name: user-api
Host: 0.0.0.0
Port: 8888
#要与目标rpc服务端依依对应
Etcd:
Hosts:
- 127.0.0.1:2379
Key: user.rpc
package config
import (
"github.com/zeromicro/go-zero/rest"
"github.com/zeromicro/go-zero/zrpc"
)
type Config struct {
rest.RestConf
//该属性保存了访问rpc服务端所需要的信息
zrpc.RpcClientConf
}
package svc
import (
"github.com/zeromicro/go-zero/zrpc"
"go_cloud_demo/rpc/types/user"
"go_cloud_demo/rpc/userclient"
"go_cloud_demo/user/internal/config"
)
type ServiceContext struct {
Config config.Config
//用来创建rpc客户端的结构体
RpcUser userclient.User
//访问rpc服务接口返回的数据
UserAuthResp *user.UserAuthResp
}
func NewServiceContext(c config.Config) *ServiceContext {
return &ServiceContext{
Config: c,
//添加初始化rpc客户端逻辑
RpcUser: userclient.NewUser(zrpc.MustNewClient(c.RpcClientConf)),
}
}
package main
import (
"context"
"flag"
"fmt"
"go_cloud_demo/rpc/types/user"
"net/http"
"go_cloud_demo/user/internal/config"
"go_cloud_demo/user/internal/handler"
"go_cloud_demo/user/internal/svc"
"github.com/zeromicro/go-zero/core/conf"
"github.com/zeromicro/go-zero/rest"
)
// 命令行参数,配置文件地址
var configFile = flag.String("f", "user/etc/user-api.yaml", "the config file")
func main() {
flag.Parse()
//1.读取配置文件解析到Config结构体上
var c config.Config
conf.MustLoad(*configFile, &c)
//2.创建服务运行上下文
ctx := svc.NewServiceContext(c)
server := rest.MustNewServer(c.RestConf)
defer server.Stop()
//3.编写访问rpc服务中Auth()接口的中间件函数
//并注册中间件
//意思是当前服务接收到请求后会先执行这个中间件函数,
//中间件内的逻辑:获取请求头中的"token"如果不存在直接响应异常
//如果存在执行ctx.RpcUser.Auth()请求rpc服务中的Auth()
//获取响应结果,如果是error直接返回,如果不是则next()向下执行
server.Use(func(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("token") == "" {
w.WriteHeader(http.StatusUnauthorized)
w.Write([]byte("Unauthorized"))
return
}
auth, err := ctx.RpcUser.Auth(context.Background(), &user.UserAuthReq{Token: r.Header.Get("token")})
if err != nil {
w.WriteHeader(http.StatusUnauthorized)
w.Write([]byte("Unauthorized"))
return
}
ctx.UserAuthResp = auth
next(w, r)
}
})
//4.注册当前服务自己的业务接口
handler.RegisterHandlers(server, ctx)
//5.监听端口启动服务
fmt.Printf("Starting server at %s:%d...\n", c.Host, c.Port)
server.Start()
}
- 微服务: 完全兼容 net/http,并且支持grpc微服务,内建服务治理逻辑,提供了服务发现,负载均衡等功能
- 高可用: 内建限流、熔断、降载，且自动触发，自动恢复
- 内建超时级联控制,自动缓存控制,链路跟踪、统计报警等
- 使用go-stash优化logstash,提高日志收集性能
- 高性能,强大的工具支持,内部整合了大量的高并发处理工具,例如MapReduce,fx数据流处理工具等等
- 支持中间件,方便扩展
- 极简的接口,基于goctl命令生成,保证代码风格
- cpu使用率到达80%触发k8s的HPA
- cpu使用率>90%时开始拒绝低优先级请求
- cpu使用率>95时开始拒绝高优先级请求
- 问题:k8s中的HAP是分钟级别的,当服务并发过高时没有触发k8s拒绝请求服务已经被打爆了
没有提供默认的重试,可以自己编写重试中间件
- 定义一个Map函数,通常情况下每个map中都存在结构,大小相同的初始化数据块,经过处理后将多个数据块转换为中间结果
- 定义一个reduce函数,通过Reduce节点接收中间数据,对中间数据进行合并,生成最终的处理结果
- 先提供一个map函数,将一个大任务分割成多个小任务,处理获取到多个中间结果
- 然后提供一个reduce函数,聚合得到的多个中间结果,拿到最终的结果
- 先提供一个map函数,将大任务分割为多个小任务,并且指定一个小任务需要几个reduce函数处理
- 然后提供一个reduce函数,在reduce函数中清洗数据进行指定梳理,处理完成后合并结果
//处理固定数量的依赖，返回error，有一个error立即返回
func Finish(fns ...func() error) error
//Finish方法功能类似，没有错误返回值
func FinishVoid(fns ...func())
func ForEach(generate GenerateFunc, mapper ForEachFunc, opts ...Option)
func MapReduce(generate GenerateFunc, mapper MapperFunc, reducer ReducerFunc, opts ...Option) (interface{}, error)
func MapReduceChan(source <-chan interface{}, mapper MapperFunc, reducer ReducerFunc,opts ...Option) (interface{}, error)
func MapReduceVoid(generate GenerateFunc, mapper MapperFunc, reducer VoidReducerFunc, opts ...Option) error
//源码:
//参数一generate: 用来生产数据的函数
//参数二mapper: 对generate生产的数据进行处理
//参数三reducer: 对mapper处理后的数据做聚合返回
func MapReduce(generate GenerateFunc, mapper MapperFunc, reducer ReducerFunc,
opts ...Option) (interface{}, error) {
panicChan := &onceChan{channel: make(chan interface{})}
source := buildSource(generate, panicChan)
return mapReduceWithPanicChan(source, panicChan, mapper, reducer, opts...)
}
//另外可能还有
//souece channel: 无缓冲的channel，用于 generte和mapper通信(generate生产的数据会写入source channel，mapper 则读取source channle的数据进行处理)
//collector channel: 有缓冲区的channe，缓冲区的长度是option.worker的数量(mapper处理完成后的数据写入collector channel，reduce读取collector数据进行处理)
//output channel: 无缓冲区channel,用于记录reducer处理后最终数据
func checkLegal(uids []int64) ([]int64, error) {
r, err := mr.MapReduce(func(source chan<- interface{}) {
// 这里是 generate | 将列表的下标值记录到 chan
// 传入到 mappe
for _, uid := range uids {
source <- uid
}
}, func(item interface{}, writer mr.Writer, cancel func(error)) {
// 这里是 mapper | 处理存入的 chan
uid := item.(int64)
ok, err := check(uid)
if err != nil {
cancel(err)
}
if ok {
writer.Write(uid)
}
}, func(pipe <-chan interface{}, writer mr.Writer, cancel func(error)) {
// 处理 reducer | 对 mapper 进行数据聚合
var uids []int64
for p := range pipe {
uids = append(uids, p.(int64))
}
// 输出到结果
writer.Write(uids)
})
if err != nil {
log.Printf("check error: %v", err)
return nil, err
}
// 对 MapReduce 结果进行转换 | 因为 MapReduce 返回的结果是 interface{} 跟咱们返回的Data类型不一致
return r.([]int64), nil
}
func check(uid int64) (bool, error) {
// do something check user legal
return true, nil
}
func productDetail(uid, pid int64) (*ProductDetail, error) {
var pd ProductDetail
err := mr.Finish(func() (err error) {
pd.User, err = userRpc.User(uid)
return
}, func() (err error) {
pd.Store, err = storeRpc.Store(pid)
return
}, func() (err error) {
pd.Order, err = orderRpc.Order(pid)
return
})
if err != nil {
log.Printf("product detail error: %v", err)
return nil, err
}
return &pd, nil
}
- mapper和reducer中都可以调用cancel，参数为error，调用后立即返回，返回结果为nil, error
- mapper中如果不调用writer.Write则item最终不会被reducer聚合
- reducer中如果不调用writer.Wirte则返回结果为nil, ErrReduceNoOutput
- reducer为单线程，所有mapper出来的结果在这里串行聚合
k8s是延迟的,采用当前
比如多线程同时创建同一个数据库链接
是logstash5倍的性能
go 书籍推荐
入门
go进阶
学习视频1
学习视频2
面试
