Go语言实战：负载均衡详解

负载均衡是分布式系统中的重要组成部分，它确保了资源的高效利用并防止请求处理中的瓶颈。在本文中，我们将通过一个简单而富有说明性的Go程序来探讨负载均衡的概念。

理解组件

在深入代码之前，让我们了解系统的关键组件：

记录器（Logger）：负责记录已处理请求的状态。
请求（Request）：表示具有唯一ID和状态标志的传入请求，标志着请求是否完成。
请求生成器（RequestGenerator）：以指定的速率生成请求，具有一定的变化性。
工作器（Worker）：模拟处理传入请求的工作器。每个工作器都有一个处理缓冲区和定义的处理时间。
负载均衡器（LoadBalancer）：使用多种负载均衡策略在工作器之间分发传入请求。

负载均衡策略

我们的负载均衡器（LoadBalancer）支持多种负载均衡策略，每种都展示了不同的处理方式：

一对一模型：每个请求都发送到一个工作器，模拟直接分配。
轮询模型：请求在工作器之间以循环顺序分发，确保工作负载均匀。
加权轮询模型：为工作器分配权重，请求根据这些权重分发。
最小连接模型：请求发送到具有最少活动连接的工作器。

Go代码

我们的Go程序包括多个组件，包括请求生成器、工作器和负载均衡器的初始化。工作器处理具有模拟处理时间的传入请求，而负载均衡器决定应该由哪个工作器处理每个传入请求。

package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

// 记录器结构
type Logger struct {
	ProcessedRequests map[int]bool
	mu                sync.Mutex
}

// 初始化记录器
func NewLogger() *Logger {
	return &Logger{
		ProcessedRequests: make(map[int]bool),
	}
}

// 记录请求处理状态
func (l *Logger) LogRequestProcessed(request *Request) {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.ProcessedRequests[request.ID] = request.Done
}

// 请求结构
type Request struct {
	ID   int
	Done bool
}

// 请求生成器
type RequestGenerator struct {
	Rate         int // 平均速率, 任务数/秒
	Variance     int // 波动率
	Duration     int // 持续时间
	RequestCount int // 请求总数
}

// 处理结构
type Worker struct {
	ID           int
	Buffer       chan *Request
	Progressed   int
	ProgressTime int
}

// 负载均衡器
type LoadBalancer struct {
	Workers []*Worker
	Current int
	Logger  *Logger
}

// 初始化请求生成器
func initializeRequestGenerator(rg *RequestGenerator, lb *LoadBalancer, wg *sync.WaitGroup) {
	defer wg.Done()

	rand.Seed(time.Now().UnixNano())
	requestGenerator := func() {
		timer := time.NewTimer(time.Duration(rg.Duration) * time.Second)
		defer timer.Stop()

		for {
			select {
			case <-timer.C:
				fmt.Println("Request generation stopped after", rg.Duration, "seconds")
				return
			default:
				request := &Request{ID: rand.Intn(1000), Done: false}
				fmt.Printf("- Sending request ID: %d\n", request.ID)
				rg.RequestCount++
				// 将请求发送给工作器
				//lb.sendRequest(request)
				//lb.sendRequestRoundRobin(request)
				lb.sendRequestLeastConnections(request)
				// 默认请求未完成
				lb.Logger.LogRequestProcessed(&Request{ID: request.ID, Done: request.Done})

				// 按照波动速率随机生成任务
				fluctuation := rand.Float64() * float64(rg.Variance)
				rate := int(float64(rg.Rate) + fluctuation)
				fmt.Println("Rate: ", rate)
				time.Sleep(time.Duration(rate) * time.Millisecond) // 控制请求发送的速率
			}
		}
	}

	go requestGenerator()
}

// 初始化工作器
func initializeWorkers(numWorkers int, lb *LoadBalancer, wg *sync.WaitGroup) {
	defer wg.Done()

	for i := 0; i < numWorkers; i++ {
		requestChan := make(chan *Request, 3) // 为每个工作器创建一个缓冲区
		worker := &Worker{
			ID:           i + 1,
			Buffer:       requestChan, // 处理能力为chan长度
			ProgressTime: i + 1,       // 每个工作器处理请求的时间
		}
		lb.Workers = append(lb.Workers, worker)

		// 启动工作器
		go worker.processRequests(lb.Logger)
	}
}

// 工作器处理请求
func (w *Worker) processRequests(logger *Logger) {
	for {
		select {
		case request := <-w.Buffer:
			fmt.Printf("-- Worker %d processing request ID: %d\n", w.ID, request.ID)

			// 请求已完成
			w.Progressed++
			request.Done = true
			logger.LogRequestProcessed(&Request{ID: request.ID, Done: request.Done})
			fmt.Printf("-- Worker %d processed time: %v\n", w.ID, w.ProgressTime)
			time.Sleep(time.Duration(w.ProgressTime) * time.Second) // 模拟处理请求的时间
		default:
			// 没有请求可用，立即结束当前循环
		}
	}
}

// 一对一模型：一个发送一个处理
func (lb *LoadBalancer) sendRequest(request *Request) {
	worker := lb.Workers[0] // 选择第一个工作器处理请求
	select {
	case worker.Buffer <- request:
		// 请求已发送
		fmt.Printf("Sending request ID: %d to worker %d\n", request.ID, worker.ID)
	default:
		// 如果通道已满，将请求标记为未处理
		fmt.Printf("Worker %d Buffer is full. Marking request ID: %d as unprocessed\n", worker.ID, request.ID)
		lb.Logger.LogRequestProcessed(&Request{ID: request.ID, Done: false})
	}
}

// 一对多模型（轮询算法）
func (lb *LoadBalancer) sendRequestRoundRobin(request *Request) {
	worker := lb.Workers[lb.Current]
	lb.Current = (lb.Current + 1) % len(lb.Workers)
	select {
	case worker.Buffer <- request:
		// 请求已发送
		fmt.Printf("Sending request ID: %d to worker %d\n", request.ID, worker.ID)
	default:
		// 如果通道已满，将请求标记为未处理
		fmt.Printf("Worker %d Buffer is full. Marking request ID: %d as unprocessed\n", worker.ID, request.ID)
		lb.Logger.LogRequestProcessed(&Request{ID: request.ID, Done: false})
	}
}

// 一对多模型（加权轮询算法）
func (lb *LoadBalancer) sendRequestWeightedRoundRobin(request *Request) {
	// 按照工作器的 ID 权重来选择
	weightedWorkers := make([]*Worker, 0)
	for _, worker := range lb.Workers {
		for i := 0; i < worker.ID; i++ {
			weightedWorkers = append(weightedWorkers, worker)
		}
	}

	// 随机选择一个工作器处理请求
	selectedWorker := weightedWorkers[rand.Intn(len(weightedWorkers))]
	selectedWorker.Buffer <- request
}

// 一对多模型（最小连接数算法）,worker的buffer长度越小，说明连接数越少,针对处理速度不一致的情况
func (lb *LoadBalancer) sendRequestLeastConnections(request *Request) {
	// 选择连接数最小的工作器处理请求
	minConnectionsWorker := lb.Workers[0]
	for _, worker := range lb.Workers {
		if len(worker.Buffer) < len(minConnectionsWorker.Buffer) {
			minConnectionsWorker = worker
		}
	}
	worker := minConnectionsWorker
	select {
	case worker.Buffer <- request:
		// 请求已发送
		fmt.Printf("Sending request ID: %d to worker %d\n", request.ID, worker.ID)
	default:
		// 如果通道已满，将请求标记为未处理
		fmt.Printf("Worker %d Buffer is full. Marking request ID: %d as unprocessed\n", worker.ID, request.ID)
		lb.Logger.LogRequestProcessed(&Request{ID: request.ID, Done: false})
	}
}

func main() {
	const (
		requestRate     = 500 // 发送请求，单位：毫秒
		requestVariance = 250 // 波动速率，单位：毫秒
		numWorkers      = 4   // 工作器数量
	)

	var wg sync.WaitGroup
	logger := NewLogger()

	rg := RequestGenerator{
		Rate:     requestRate,
		Variance: requestVariance,
		Duration: 30,
	}

	loadBalancer := LoadBalancer{
		Workers: make([]*Worker, 0),
		Current: 0,
		Logger:  logger,
	}

	// 初始化请求生成器
	wg.Add(1)
	go initializeRequestGenerator(&rg, &loadBalancer, &wg)

	// 初始化工作器
	wg.Add(1)
	go initializeWorkers(numWorkers, &loadBalancer, &wg)

	wg.Wait()

	time.Sleep(time.Second * 40)
	var Unprocessed int
	Unprocessed = rg.RequestCount

	// 打印请求处理状态
	fmt.Println("Request Count: ", rg.RequestCount)

	// 打印工作器处理的请求数量
	for _, worker := range loadBalancer.Workers {
		Unprocessed -= worker.Progressed
		fmt.Printf("Worker %d processed %d requests\n", worker.ID, worker.Progressed)
	}

	// 打印请求处理状态
	fmt.Println("Request Processed:", logger.ProcessedRequests)

	// 打印请求未处理条数
	fmt.Println("Request Unprocessed:", Unprocessed)
}

展示负载均衡

为了展示我们负载均衡策略的有效性，我们设置了一个具有一定请求速率、变异性和工作器数量的场景。在运行模拟之后，我们分析已处理请求、每个工作器的工作负载以及整个系统的效率。

结论

负载均衡是分布式系统中的基本概念，它确保了资源的最佳利用并防止性能瓶颈。在本文中，我们探讨了一个简单的Go程序，实现了不同的负载均衡策略，演示了不同方法在工作器之间如何分配工作负载。

随时尝试修改代码，调整参数，并观察系统在不同条件下的行为。了解负载均衡对于设计可伸缩且可靠的分布式系统至关重要。

Happy coding!

hypo

https://github.com/hypo-Z/hypo-Z.github.io/2023/12/21/go-yu-yan-shi-zhan-fu-zai-jun-heng-xiang-jie/

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hypo !

Golang 负载均衡

Go语言实用——好用也有三方库之lo

好用也有三方库之lo

2024-01-10 笔记

笔记 Golang 实用

如何更好的理解闭包的原理

在Go语言中，闭包（closures）是指一个函数值（function value）捕获了其所在作用域外部变量的引用。

2023-10-23 笔记

Golang 闭包