当前位置：网站首页>17、负载均衡

17、负载均衡

2022-08-01 19:22:00 【无休止符】

一、动态获取端口

在实现负载均衡之前，我们还需要进行一个小优化
因为在使用负载均衡的时候会启动相同服务的多个实例，之前我们都是将端口配置在yaml中
如果多个服务启动的时候还使用端口配置的方案，会导致端口冲突
所以我们需要先进行优化，可以动态获取可用端口

1 - user_web层优化

user_web/utils/addr.go：添加获取动态端口的方法

package utils

import (
	"net"
)

func GetFreePort() (int, error) {
    
	addr, err := net.ResolveTCPAddr("tcp", "localhost:0")
	if err != nil {
    
		return 0, err
	}

	l, err := net.ListenTCP("tcp", addr)
	if err != nil {
    
		return 0, err
	}
	defer l.Close()
	return l.Addr().(*net.TCPAddr).Port, nil
}

user_web/main.go：调用动态化端口获取

package main

import (
	"fmt"
	"github.com/spf13/viper"
	"web_api/user_web/global"
	"web_api/user_web/initialize"
	"web_api/user_web/utils"

	"github.com/gin-gonic/gin/binding"
	ut "github.com/go-playground/universal-translator"
	"github.com/go-playground/validator/v10"
	"go.uber.org/zap"

	myvalidator "web_api/user_web/validator"
)

func main() {
    
	//1. 初始化logger
	initialize.InitLogger()
	//2. 初始化配置文件
	initialize.InitConfig()
	//3. 初始化routers
	Router := initialize.Routers()
	//4. 初始化翻译
	if err := initialize.InitTrans("zh"); err != nil {
    
		panic(err)
	}

	//5. 初始化srv的连接
	initialize.InitSrvConn()

	viper.AutomaticEnv()
	//如果是本地开发环境端口号固定，线上环境启动获取端口号
	debug := viper.GetBool("DEV_CONFIG")
	if !debug {
    
		port, err := utils.GetFreePort()
		if err == nil {
    
			global.ServerConfig.Port = port
		}
	}

	//。。。省略
}

2 - user_srv层优化

user_srv/utils/addr.go：添加获取动态端口的方法

package utils

import (
	"net"
)

func GetFreePort() (int, error) {
    
	addr, err := net.ResolveTCPAddr("tcp", "localhost:0")
	if err != nil {
    
		return 0, err
	}

	l, err := net.ListenTCP("tcp", addr)
	if err != nil {
    
		return 0, err
	}
	defer l.Close()
	return l.Addr().(*net.TCPAddr).Port, nil
}

user_srv/main.go：
- 默认启动端口为0，如果我们从命令行带参数启动的话就不会为0
- 是否动态获取端口就判断port是否为0，为0就动态获取端口

package main

import (
	"flag"
	"fmt"
	"github.com/hashicorp/consul/api"
	"go.uber.org/zap"
	"google.golang.org/grpc/health"
	"google.golang.org/grpc/health/grpc_health_v1"
	"nd/user_srv/global"
	"nd/user_srv/handler"
	"nd/user_srv/initialize"
	"nd/user_srv/proto"
	"nd/user_srv/utils"
	"net"

	"google.golang.org/grpc"
)

func main() {
    
	IP := flag.String("ip", "0.0.0.0", "ip地址")
	Port := flag.Int("port", 0, "端口号") // 这个修改为0，如果我们从命令行带参数启动的话就不会为0

	//初始化
	initialize.InitLogger()
	initialize.InitConfig()
	initialize.InitDB()
	zap.S().Info(global.ServerConfig)

	flag.Parse()
	zap.S().Info("ip: ", *IP)
	if *Port == 0 {
    
		*Port, _ = utils.GetFreePort()
	}
	zap.S().Info("port: ", *Port)

	//。。。省略
}

二、负载均衡简介

1 - 什么是负载均衡

微服务中用户的请求过程：
- ①.用户请求通过nginx实现负载均衡到达网关
- ②.网关再通过负载均衡将请求分配到不同的用户web服务
- ③.用户web服务通过grpc，负载均衡分配到不同的用户srv服务调用请求

在这里插入图片描述

2 - 负载均衡策略

a、集中式LB：即在服务的消费方和提供方之间使用独立的LB设施(可以是硬件，如F5, 也可以是软件，如nginx),由该设施负责把访问请求通过某种策略转发至服务的提供方
- 服务消费方如何发现LB呢？通常的做法是通过DNS，运维人员为服务配置一个DNS域名，这个域名指向LB。这种方案基本可以否决，因为它有致命的缺点：所有服务调用流量都经过load balance服务器，所以load balace服务器成了系统的单点，一旦LB发生故障对整个系统的影响是灾难性的。
- 为了解决这个问题，必然需要对这个load balance部件做分布式处理（部署多个实例、冗余，然后解决一致性问题等全家桶解决方案），但这样做会徒增非常多的复杂度
b、进程内LB：将LB逻辑集成到消费方，消费方从服务注册中心获知有哪些地址可用，然后自己再从这些地址中选择出一个合适的服务器。
调用过程
- 1、用户-srv服务注册到consul注册中心中
- 2、用户web层通过goroutine定期向consul拉取user_srv的ip和端口，维护到服务列表中
- 3、服务列表中保存的是与user_srv服务的连接（不是ip和port，这样可以减少三次握手）
- 4、用户调用接口的时候通过用户web层向服务列表请求，服务列表根据LB算法分配本次请求的连接
c、独立进程LB：该方案是针对第二种方案的不足提出的一种折中方案，原理和第二种方案基本类似，不同之处是，它将LB和服务发现功能从进程内移出来，变成主机上的一个独立进程，主机上的一个或者多个服务要访问目标服务时，他们都通过同一主机上的独立LB进程做服务发现和负载均衡（简单的理解就是，将LB也部署成服务）
- 缺点：这个方案解决上一种方案的问题，不需要为不同语言开发用户库，LB升级不需要服务调用方修改代码；但引入新的问题是 —— 这个组件本身的可用性谁来维护？还要再写一个watchdog去监控这个组件吗？另外，多了一个环节，就多了一个出错的可能，线上出问题了，也多了一个需要维护的环节

在这里插入图片描述

三、常用负载均衡算法

轮询法（Round Robin）：轮询很容易实现，将请求按顺序轮流分配到后台服务器上，均衡的对待每一台服务器，而不关心服务器实际的连接数和当前的系统负载（缺点是，如果服务器的配置不同，比如1台配置是4核8G、8核16G、16核64G）
随机法：通过系统随机函数，根据后台服务器列表的大小值来随机选取其中一台进行访问。由概率统计理论可以得知，随着调用量的增大，其实际效果越来越接近于平均分配流量到后台的每一个服务器，也就是轮询法的效果
源地址哈希法：源地址哈希法的思想是根据服务消费者请求客户端的ip地址，通过哈希函数计算得到一个哈希值，将此哈希值和服务器列表的大小进行取模运算，得到的结果便是要访问的服务器地址的序号；采用源地址哈希法进行负载均衡，相同的IP客户端，如果服务器列表不变，将映射到同一后台服务器进行访问
加权轮询法：不同的后台服务器可能机器的配置和当前系统的负载并不相同，因此他们的抗压能力也不一样。跟配置高、负载低的机器分配更高的权重，使其能处理更多的请求，而配置低、负载高的机器，则给其分配较低的权重，降低其系统负载，加权轮询很好的处理了这一问题，并将请求按照顺序且根据权重分配给后端
加权随机法（weight random）：加权随机法跟加权轮询法类似，根据后台服务器不同的配置和负载情况，配置不同的权重。不同的是，它是按照权重来随机选取服务器的，而非顺序
最小连接数法：前面我们费尽心思来实现服务消费者请求次数分配的均衡，我们知道这样做是没错的，可以为后端的多台服务器平均分配工作量，最大程度地提高服务器的利用率，但是，实际上，请求次数的均衡并不代表负载的均衡。因此我们需要介绍最小连接数法，最小连接数法比较灵活和智能，由于后台服务器的配置不尽相同，对请求的处理有快有慢，它正是根据后端服务器当前的连接情况，动态的选取其中当前积压连接数最少的一台服务器来处理当前请求，尽可能的提高后台服务器利用率，将负载合理的分流到每一台服务器

四、grpc负载均衡策略

grpc官方负载均衡doc：https://github.com/grpc/grpc/blob/master/doc/load-balancing.md
grpc的consul解决方案：https://github.com/mbobakov/grpc-consul-resolver

五、grpc实现负载均衡

需要解决的2个问题
- 如何启动两个服务
- 即使能通过终端启动两个服务，但是注册到consul中的时候也会被覆盖
解决方案
- 服务使用终端启动
- 注册consul的时候，使用相同的registration.Name，但是不同的registration.ID（ID使用uuid来随机生成）
user_srv/main.go
- 为了让服务退出的时候，consul就注销掉对应的服务，而不是等待consul注销：使用go程实现监听，并添加终止信号的逻辑

package main

import (
	"flag"
	"fmt"
	"github.com/hashicorp/consul/api"
	"github.com/satori/go.uuid"
	"go.uber.org/zap"
	"google.golang.org/grpc/health"
	"google.golang.org/grpc/health/grpc_health_v1"
	"nd/user_srv/global"
	"nd/user_srv/handler"
	"nd/user_srv/initialize"
	"nd/user_srv/proto"
	"nd/user_srv/utils"
	"net"
	"os"
	"os/signal"
	"syscall"

	"google.golang.org/grpc"
)

func main() {
    
	IP := flag.String("ip", "0.0.0.0", "ip地址")
	Port := flag.Int("port", 0, "端口号") // 这个修改为0，如果我们从命令行带参数启动的话就不会为0

	//初始化
	initialize.InitLogger()
	initialize.InitConfig()
	initialize.InitDB()
	zap.S().Info(global.ServerConfig)

	flag.Parse()
	zap.S().Info("ip: ", *IP)
	if *Port == 0 {
    
		*Port, _ = utils.GetFreePort()
	}
	zap.S().Info("port: ", *Port)

	server := grpc.NewServer()
	proto.RegisterUserServer(server, &handler.UserServer{
    })
	lis, err := net.Listen("tcp", fmt.Sprintf("%s:%d", *IP, *Port))
	if err != nil {
    
		panic("failed to listen:" + err.Error())
	}

	//注册服务健康检查
	grpc_health_v1.RegisterHealthServer(server, health.NewServer())

	//服务注册
	cfg := api.DefaultConfig()
	cfg.Address = fmt.Sprintf("%s:%d", global.ServerConfig.ConsulInfo.Host,
		global.ServerConfig.ConsulInfo.Port)

	client, err := api.NewClient(cfg)
	if err != nil {
    
		panic(err)
	}
	//生成对应的检查对象
	check := &api.AgentServiceCheck{
    
		GRPC:                           fmt.Sprintf("192.168.91.1:%d", *Port),
		Timeout:                        "5s",
		Interval:                       "5s",
		DeregisterCriticalServiceAfter: "15s",
	}

	//生成注册对象
	registration := new(api.AgentServiceRegistration)
	registration.Name = global.ServerConfig.Name
	serviceID := fmt.Sprintf("%s", uuid.NewV4())
	registration.ID = serviceID
	registration.Port = *Port
	registration.Tags = []string{
    "imooc", "bobby", "user", "srv"}
	registration.Address = "192.168.91.1"
	registration.Check = check

	err = client.Agent().ServiceRegister(registration)
	if err != nil {
    
		panic(err)
	}

	go func() {
    
		err = server.Serve(lis)
		if err != nil {
    
			panic("failed to start grpc:" + err.Error())
		}
	}()

	//接收终止信号
	quit := make(chan os.Signal)
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	<-quit
	if err = client.Agent().ServiceDeregister(serviceID); err != nil {
    
		zap.S().Info("注销失败")
	}
	zap.S().Info("注销成功")
}

user_srv/handler/user.go：在GetUserList添加打印日志

func (s *UserServer) GetUserList(ctx context.Context, req *proto.PageInfo) (*proto.UserListResponse, error) {
    
	//获取用户列表
	var users []model.User
	result := global.DB.Find(&users)
	if result.Error != nil {
    
		return nil, result.Error
	}
	fmt.Println(time.Now().Format("2006-01-02 15:04:05.000"), "用户列表")
	rsp := &proto.UserListResponse{
    }
	rsp.Total = int32(result.RowsAffected)

	global.DB.Scopes(Paginate(int(req.Pn), int(req.PSize))).Find(&users)

	for _, user := range users {
    
		userInfoRsp := ModelToResponse(user)
		rsp.Data = append(rsp.Data, &userInfoRsp)
	}
	return rsp, nil
}

测试脚本：用来测试负载均衡：这里要把user的proto拷贝过来

package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc/credentials/insecure"
	"log"
	"test/proto"

	_ "github.com/mbobakov/grpc-consul-resolver" // It's important

	"google.golang.org/grpc"
)

func main() {
    
	conn, err := grpc.Dial(
		"consul://192.168.91.129:8500/user_srv?wait=14s&tag=srv",
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy": "round_robin"}`),
	)
	if err != nil {
    
		log.Fatal(err)
	}
	defer conn.Close()

	for i := 0; i < 10; i++ {
    
		userSrvClient := proto.NewUserClient(conn)
		rsp, err := userSrvClient.GetUserList(context.Background(), &proto.PageInfo{
    
			Pn:    1,
			PSize: 2,
		})
		if err != nil {
    
			panic(err)
		}
		for index, data := range rsp.Data {
    
			fmt.Println(index, data)
		}
	}
}

在这里插入图片描述

终端1：5次打印 —— “用户列表”
终端2：5次打印 —— “用户列表”

在这里插入图片描述

六、gin集成负载均衡

user_web/initialize/init_srv_conn.go：修改之前初始化连接的方法

package initialize

import (
	"fmt"
	"github.com/hashicorp/consul/api"
	_ "github.com/mbobakov/grpc-consul-resolver" // It's important
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"web_api/user_web/global"
	"web_api/user_web/proto"
)

func InitSrvConn() {
    
	consulInfo := global.ServerConfig.ConsulInfo
	userConn, err := grpc.Dial(
		fmt.Sprintf("consul://%s:%d/%s?wait=14s", consulInfo.Host, consulInfo.Port, global.ServerConfig.UserSrvInfo.Name),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy": "round_robin"}`),
	)
	if err != nil {
    
		zap.S().Fatal("[InitSrvConn] 连接 【用户服务失败】")
	}

	userSrvClient := proto.NewUserClient(userConn)
	global.UserSrvClient = userSrvClient
}

测试方法
- golang启动user_web项目
- 终端启动2个user_srv项目
- YApi发送查询用户列表请求
- 可以看到2个终端的输出是交替打印的（负载均衡）

在这里插入图片描述

七、完整源码

mxshop_srvsV7.0.rar

附录
- user_srv/config_debug.yaml：mysql的host、consul的host需要自行修改
- user_web/config_debug.yaml：redis的host、consul的host需要自行修改
- user_srv/main.go：健康检查与consul注册的ip地址需要修改
- user_srv/tests/user.go：如要使用TestGetUserList与TestCreateUser需要修改端口号

在这里插入图片描述