golang实现负载均衡算法
1、真实服务器
package mainimport ( 'fmt' 'log' 'net/http' 'os' 'os/signal' 'strconv' 'syscall' 'time')type realServer struct { Addr string}func (rs *realServer) HelloHandler(w http.ResponseWriter,r *http.Request){ data := fmt.Sprintf('[%s] http://%s%s nn',rs.Addr,rs.Addr,r.RequestURI) w.Write([]byte(data))}func (rs *realServer) Run(){ fmt.Println('Http server tart to serve at :',rs.Addr) mux := http.NewServeMux() mux.HandleFunc('/',rs.HelloHandler) server := &http.Server{ Addr: rs.Addr, Handler: mux, WriteTimeout: time.Second * 3, } go func(){ if err := server.ListenAndServe();err != nil{ log.Fatal('Start http server failed,err:',err) } }()}func main() { doneCh := make(chan os.Signal) for i:=0;i<5;i++{ port := '808' + strconv.Itoa(i) addr := '127.0.0.1:' + port rs := &realServer{Addr: addr} go rs.Run() } signal.Notify(doneCh,syscall.SIGINT,syscall.SIGTERM) <- doneCh}2、反向代理代码框架
// package httpServerimport ( 'math/rand' 'time')type HttpServer struct { Host string}type LoadBalance struct { Servers []*HttpServer}func NewLoadBalance()*LoadBalance{ return &LoadBalance{Servers:make([]*HttpServer,0)}}func NewHttpServer(host string)*HttpServer{ return &HttpServer{ Host:host, }}func (lb *LoadBalance)Add(server *HttpServer){ lb.Servers = append(lb.Servers,server)}启动服务
// server.gopackage mainimport ( 'log' 'net/http' 'net/http/httputil' 'net/url' . 'gostudy/reverseProxyDemo/httpServer')type ReveseProxyHandler struct {}func (rph *ReveseProxyHandler)ServeHTTP(w http.ResponseWriter,r *http.Request){ lb := NewLoadBalance() lb.Add(NewHttpServer('http://127.0.0.1:8080')) lb.Add(NewHttpServer('http://127.0.0.1:8081')) lb.Add(NewHttpServer('http://127.0.0.1:8082')) lb.Add(NewHttpServer('http://127.0.0.1:8083')) lb.Add(NewHttpServer('http://127.0.0.1:8084')) url,err := url.Parse(lb.GetHttpServerByRandom().Host) if err != nil { log.Println('[ERR] url.Parse failed,err:',err) return } proxy := httputil.NewSingleHostReverseProxy(url) proxy.ServeHTTP(w,r)}func main() { proxy := &ReveseProxyHandler{} log.Println('Start to serve at 127.0.0.1:8888') if err := http.ListenAndServe(':8888',proxy);err !=nil{ log.Fatal('Failed to start reverse proxy server ,err:',err) }}3、随机负载均衡算法
// httpServer/reverseProxy.go// 随机负载均衡func (lb *LoadBalance)GetHttpServerByRandom()*HttpServer{ rand.Seed(time.Now().UnixNano()) index := rand.Intn(len(lb.Servers)) return lb.Servers[index]}测试结果
$ for i in {0..9};do curl -s http://127.0.0.1:8888/reverseproxydemo?id=123;done[127.0.0.1:8083] http://127.0.0.1:8083/reverseproxydemo?id=123[127.0.0.1:8084] http://127.0.0.1:8084/reverseproxydemo?id=123[127.0.0.1:8082] http://127.0.0.1:8082/reverseproxydemo?id=123[127.0.0.1:8080] http://127.0.0.1:8080/reverseproxydemo?id=123[127.0.0.1:8081] http://127.0.0.1:8081/reverseproxydemo?id=123[127.0.0.1:8082] http://127.0.0.1:8082/reverseproxydemo?id=123[127.0.0.1:8080] http://127.0.0.1:8080/reverseproxydemo?id=123[127.0.0.1:8081] http://127.0.0.1:8081/reverseproxydemo?id=123[127.0.0.1:8080] http://127.0.0.1:8080/reverseproxydemo?id=123[127.0.0.1:8084] http://127.0.0.1:8084/reverseproxydemo?id=123加权随机
原理：获取到所有节点的权重值，将weight个当前节点Index加到一个[]int，并随机从中获取一个index，例如：A ： B ： C = 5:2:1 且ABC三个节点的Index分别为0,1,2，那么新建一个如下是切片：[]int{0,0,0,0,0,1,1,2} ，然后通过rand(len([]int)) 随机拿到一个index
// httpserver.gotype HttpServer struct { Host string Weight int}func NewHttpServer(host string,weight int)*HttpServer{ return &HttpServer{ Host:host, Weight:weight, }}// 加权随机func (lb *LoadBalance)GetHttpServerByRandomWithWeight(httpServerArr []int)*HttpServer{ rand.Seed(time.Now().UnixNano()) index := rand.Intn(len(httpServerArr)) return lb.Servers[httpServerArr[index]]}// loadBalanceDemo/loadbalance.go// 加权随机 var httpServerArr []int for index,server := range lb.Servers{ if server.Weight > 0 { for i:=0;i<server.Weight;i++{ httpServerArr = append(httpServerArr,index) } } } url,err := url.Parse(lb.GetHttpServerByRandomWithWeight(httpServerArr).Host)加权随机算法优化版
上面的加权随机算法实现起来比较简单，但存在一个明显弊端，如果weight值的大小将直接影响切片大小，例如5:2 跟 50000:20000 本质上是一样的，但后者将占用更多的内存空间。因此我们需要对该算法做下优化，将N个节点权重计算出N个区间，然后取随机数rand(weightSum)，看该数落在哪个区间就返回该区间对应的index值，举个例子：假设A:B:C = 5:2:1 那么我们先计算出3个区间：5,7(5+2),8(5+2+1)[0,5) [5,7) [7,8)然后取rand(5+2+1)，假设获取到的值为6，则落在[5,7) 这个区间，返回index=1可以看出rand(7)随机数落在各个区间分布如下：[0,5) ： 0,1,2,3,4[5,7) ：5,6[7,8) ：7正好是5:2:1
下面是具体实现：
// 加权随机优化版func (lb *LoadBalance)GetHttpServerByRandomWithWeight2()*HttpServer{ rand.Seed(time.Now().UnixNano()) // 计算所有节点权重值之和 weightSum := 0 for i:=0;i<len(lb.Servers);i++{ weightSum += lb.Servers[i].Weight } // 随机数获取 randNum := rand.Intn(weightSum) sum := 0 for i := 0;i<len(lb.Servers);i++{ sum += lb.Servers[i].Weight // 因为区间是[ ) ，左闭右开，故随机数小于当前权重sum值，则代表落在该区间，返回当前的index if randNum < sum { return lb.Servers[i] } } return lb.Servers[0]}轮询算法
假设有ABC 3台机器，那么请求过来将按照ABCABC 这样的顺顺序将请求反向代理到后端服务器
原理是记录当前的index值，每次请求+1 取模（这里仅演示算法，未考虑线程安全问题，没有加锁）
// loadbalance.go// 由于每次请求需要保存当前的index值，所以使用全局变量lb，并在初始化函数中初始化lb实例var lb *LoadBalancefunc init(){ lb = NewLoadBalance()}// httpserver.go// 结构体中加上当前index值type LoadBalance struct { Index int Servers []*HttpServer}// 轮询func (lb *LoadBalance)GetHttpServerByRoundRobin() *HttpServer{ server := lb.Servers[lb.Index] lb.Index = (lb.Index + 1)% len(lb.Servers) return server}加权轮询算法-切片算法
/ 加权轮询func (lb *LoadBalance)GetHttpServerByRoundRobinWithWeight(indexArr []int) *HttpServer{ lb.Index = (lb.Index + 1)% len(indexArr) fmt.Println(indexArr) return lb.Servers[indexArr[lb.Index]]}package mainimport ( 'log' 'net/http' 'net/http/httputil' 'net/url' . 'loadBalanceDemo/httpServer')type ReveseProxyHandler struct {}var lb *LoadBalancevar indexArr []intfunc init(){ lb = NewLoadBalance() lb.Add(NewHttpServer('http://127.0.0.1:8082',5)) lb.Add(NewHttpServer('http://127.0.0.1:8083',2)) lb.Add(NewHttpServer('http://127.0.0.1:8084',1)) // 加权轮询 indexArr = make([]int,0) for index,server := range lb.Servers{ if server.Weight > 0{ for i:=0;i<server.Weight;i++{ indexArr = append(indexArr,index) } } }}func (rph *ReveseProxyHandler)ServeHTTP(w http.ResponseWriter,r *http.Request){ // 浏览器访问时默认会请求/favicon.ico，这里忽略该URL if r.URL.Path == '/favicon.ico'{ return } url,err := url.Parse(lb.GetHttpServerByRoundRobinWithWeight(indexArr).Host) if err != nil { log.Println('[ERR] url.Parse failed,err:',err) return } proxy := httputil.NewSingleHostReverseProxy(url) proxy.ServeHTTP(w,r)}func main() { proxy := &ReveseProxyHandler{} log.Println('Start to serve at 127.0.0.1:8888') if err := http.ListenAndServe(':8888',proxy);err !=nil{ log.Fatal('Failed to start reverse proxy server ,err:',err) }}加权轮询算法-区间算法
// 加权轮询区间算法func (lb *LoadBalance)GetHttpServerByRoundRobinWithWeight2()*HttpServer{ server := lb.Servers[0] sum := 0 for i:=0;i<len(lb.Servers);i++{ sum += lb.Servers[i].Weight if lb.Index < sum{ server = lb.Servers[i] if lb.Index == sum -1 && i != len(lb.Servers)-1{ lb.Index++ }else{ lb.Index = (lb.Index+1) % sum } fmt.Println(lb.Index) break } } return server}ip_hash 算法
// ip_hash// 对客户端IP 做hash 取模得到有一个固定的index，返回固定的httpserverfunc (lb *LoadBalance)GetHttpServerByIpHash(ip string) *HttpServer{ index := int(crc32.ChecksumIEEE([]byte(ip))) % len(lb.Servers) return lb.Servers[index]}// server.go// ip_hash// 传入客户端IP url,err := url.Parse(lb.GetHttpServerByIpHash(r.RemoteAddr).Host)url_hash 算法
// url_hash url,err := url.Parse(lb.GetHttpServerByUrlHash(r.RequestURI).Host)// url_hashfunc (lb *LoadBalance) GetHttpServerByUrlHash(url string) *HttpServer{ index := int(crc32.ChecksumIEEE([]byte(url))) % len(lb.Servers) return lb.Servers[index]}