使用加权轮询算法和 Go 实现 HTTP 负载分发代理

Blog Logo
on



本篇主要从以下两个方面进行展开：

  1. 使用 Go 实现一个反向代理
  2. 使用 WRR 算法实现此反向代理的加权轮询

什么是代理

代理一般情况下我们是分为正向代理和反向代理两种形式

正向代理一般是配置在客户端，客户端需要知道代理的地址并自行配置

反向代理一般情况下对客户端是透明的，主要是配置在服务器上，大部分我们访问的 Web 应用都是通过反向代理进行配置的，这块主要是 Nginx 和 Apache 提供的功能，比如 Nginx：

location / {
    proxy_pass: http://127.0.0.1:3000
}

还有一个简单区分的方式，我们可以看这个代理的作用，如果已知转发的后端，那这个应该是反向代理，如果被转发到哪里去是一个不确定性因素，那这应该是一个正向代理，这是我的个人理解，欢迎读者批评指正 :)

Go ReverseProxy

Go 提供了一个 httputil.ReverseProxy 代理框架，能够让我们快速的实现一个反向代理而无需关注其他细节。

打开reverseproxy.go源码可以看到对 ReverseProxy 的简介

// ReverseProxy is an HTTP Handler that takes an incoming request and
// sends it to another server, proxying the response back to the
// client.
//
// ReverseProxy automatically sets the client IP as the value of the
// X-Forwarded-For header.
// If an X-Forwarded-For header already exists, the client IP is
// appended to the existing values.
// To prevent IP spoofing, be sure to delete any pre-existing
// X-Forwarded-For header coming from the client or
// an untrusted proxy.

ReverseProxy 是一个用来转发请求的服务，并把相关的响应原封不动的返回给客户端，这个过程中 ReverseProxy 会自动给X-Forwarded-For添加或者追加客户端 IP

继续往下看

type ReverseProxy struct {
	// Director must be a function which modifies
	// the request into a new request to be sent
	// using Transport. Its response is then copied
	// back to the original client unmodified.
	// Director must not access the provided Request
	// after returning.
	Director func(*http.Request)

	// The transport used to perform proxy requests.
	// If nil, http.DefaultTransport is used.
	Transport http.RoundTripper
...

ReverseProxy 是一个结构体，其中最重要的是 Director，我们需要实现一个 Director 函数，这个函数主要就是让我们可以定义对request的修改，比如修改协议、回源地址、路径等信息，然后传递给http.RoundTripper进行转发，RoundTripper 主要的功能就是返回响应给一个给定的请求，具体可以查看client.go源码，这里不再发散

一个简单的反向代理服务

下面是使用 ReverseProxy 实现的一个简单的反向代理服务：

package main

import (
	"fmt"
	"net/http"
	"net/http/httputil"
	"net/url"
)

func main() {
	// generateBackend
	node := url.URL{Host: "127.0.0.1:7791", Scheme: "http"}

	// generate a reverse proxy
	reverseProxy := httputil.NewSingleHostReverseProxy(&node)

	fmt.Println("Server started on port 7788...")
	if err := http.ListenAndServe("127.0.0.1:7788", reverseProxy); err != nil {
		fmt.Printf("Server failed to start, error: %s \n", err.Error())
	}
}

我们启动此服务，并且使用我自己写的一个testServer.go进行测试

// generate 3 server to handle request for test

package main

import (
	"fmt"
	"net/http"
)

type testServer struct {
	addr         string
	requestCount int
}

func (server *testServer) handler(res http.ResponseWriter, req *http.Request) {
	server.requestCount += 1
	fmt.Printf("%s: %s %s on serverHandlerPort: %s, Total: %d\n",
		req.Proto, req.Method, req.RequestURI, server.addr, server.requestCount)
	res.Write([]byte(server.addr))
}

func main() {
	nodeA := testServer{"127.0.0.1:7791", 0}
	nodeB := testServer{"127.0.0.1:7792", 0}
	nodeC := testServer{"127.0.0.1:7793", 0}

	nodes := []testServer{nodeA, nodeB, nodeC}

	serverA := http.NewServeMux()
	serverA.HandleFunc("/", nodeA.handler)
	go http.ListenAndServe(nodeA.addr, serverA)

	serverB := http.NewServeMux()
	serverB.HandleFunc("/", nodeB.handler)
	go http.ListenAndServe(nodeB.addr, serverB)

	serverC := http.NewServeMux()
	serverC.HandleFunc("/", nodeC.handler)
	go http.ListenAndServe(nodeC.addr, serverC)

	var comm int

	for _, node := range nodes {
		fmt.Printf("Server %s started...\n", node.addr)
	}

	fmt.Println("All server are started, input any key and enter to quit...")
	fmt.Scan(&comm)
	fmt.Printf("Result: ServerA: %d  ServerB: %d  ServerC: %d\n",
		nodeA.requestCount, nodeB.requestCount, nodeC.requestCount)
}

使用命令访问curl 127.0.0.1:7788，可以看到已经将请求转发到后端127.0.0.1:7791了

输入图片说明

以上，我们就已经实现了一个反向代理，用来对请求进行代理，但这并不是我们想要的，我们想要的是分发，所以我们继续改造此脚本，使它能够配置多个后端进行轮询

支持多个后端轮询的反向代理服务

参照源码中NewSingleHostReverseProxy函数自定义即可

package main

import (
	"fmt"
	"math/rand"
	"net/http"
	"net/http/httputil"
	"net/url"
)

//func NewSingleHostReverseProxy(target *url.URL) *ReverseProxy {
//	targetQuery := target.RawQuery
//	director := func(req *http.Request) {
//		req.URL.Scheme = target.Scheme
//		req.URL.Host = target.Host
//		req.URL.Path = singleJoiningSlash(target.Path, req.URL.Path)
//		if targetQuery == "" || req.URL.RawQuery == "" {
//			req.URL.RawQuery = targetQuery + req.URL.RawQuery
//		} else {
//			req.URL.RawQuery = targetQuery + "&" + req.URL.RawQuery
//		}
//		if _, ok := req.Header["User-Agent"]; !ok {
//			// explicitly disable User-Agent so it's not set to default value
//			req.Header.Set("User-Agent", "")
//		}
//	}
//	return &ReverseProxy{Director: director}
//}

// config for backends
var nodes = [] *url.URL{
	{Host: "127.0.0.1:7791", Scheme: "http"},
	{Host: "127.0.0.1:7792", Scheme: "http"},
	{Host: "127.0.0.1:7793", Scheme: "http"},
}

func generateReverseProxy() *httputil.ReverseProxy{
	// generate director
	director := func(req *http.Request) {
		backend := nodes[rand.Int()%len(nodes)]
		req.URL.Scheme = backend.Scheme
		req.URL.Host = backend.Host
		fmt.Printf("Scheme: %s Host: %s Path: %s\n", req.Proto, req.Host, req.RequestURI)
	}
	return &httputil.ReverseProxy{Director: director}
}

func main() {
	// generate a reverse proxy
	reverseProxy := generateReverseProxy()

	fmt.Println("Server started on port 7788...")
	if err := http.ListenAndServe("127.0.0.1:7788", reverseProxy); err != nil {
		fmt.Printf("Server failed to start, error: %s \n", err.Error())
	}
}

结果如下图

输入图片说明

一共访问40次，分布均匀性一般般，所以接下来就是改造为加权轮询算法进行后端服务器的选择

加权轮询算法（Weighted Round-Robin）

加权轮询算法的实现可以参见 Nginx 的一次代码提交 Upstream: smooth weighted round-robin balancing.

简单来说，Nginx 的这个加权轮询算法不仅仅可以实现按照权重进行分发负载，也实现了平滑性，什么叫平滑呢，就是说你设置了这么一组带权重的后端

	{HostA: "127.0.0.1:7791", Weight: 3},
	{HostB: "127.0.0.1:7792", Weight: 1},
	{HostC: "127.0.0.1:7793", Weight: 1}

比如这时来了5次访问，那么它的分发如果是AAABC，我们就不能说这个是「平滑」的，因为可能会给第一台机器造成压力过大，虽然我们认为它的性能比较好，但是还是会造成同一时间压力过大的问题。而 WRR 算法就可以实现平滑的分发，使分发变成ABACA，避免同一时间造成压力过大的问题，来看看这个算法如何实现的：

For edge case weights like { 5, 1, 1 } we now produce { a, a, b, a, c, a, a }
sequence instead of { c, b, a, a, a, a, a } produced previously.

Algorithm is as follows: on each peer selection we increase current_weight
of each eligible peer by its weight, select peer with greatest current_weight
and reduce its current_weight by total number of weight points distributed
among peers.

In case of { 5, 1, 1 } weights this gives the following sequence of
current_weight's:

     a  b  c
     0  0  0  (initial state)

     5  1  1  (a selected)
    -2  1  1

     3  2  2  (a selected)
    -4  2  2

     1  3  3  (b selected)
     1 -4  3

     6 -3  4  (a selected)
    -1 -3  4

     4 -2  5  (c selected)
     4 -2 -2

     9 -1 -1  (a selected)
     2 -1 -1

     7  0  0  (a selected)
     0  0  0

To preserve weight reduction in case of failures the effective_weight
variable was introduced, which usually matches peer's weight, but is
reduced temporarily on peer failures.

简单来说就是：

  1. 每一轮选择都用自身的权重加到当前权重
  2. 当前选中的节点的当前权重需要减去总权重

先来看看算法用 Go 的实现

package main

import "fmt"

type wrrServer struct {
	address       string
	weight        int
	currentWeight int
}

// nginx weighted round-robin balancing
// view: https://github.com/phusion/nginx/commit/27e94984486058d73157038f7950a0a36ecc6e35

func getBestServer(servers []*wrrServer) (b *wrrServer) {
	allWeight := 0
	for _, server := range servers {
		if server == nil {
			return nil
		}

		allWeight += server.weight            // 计算总权重
		server.currentWeight += server.weight // 当前权重加上权重

		if b == nil || server.currentWeight > b.currentWeight { // 如果最优节点不存在或者当前节点由于最优节点，则赋值或者替换
			b = server
		}
	}

	if b == nil {
		return nil
	}

	b.currentWeight -= allWeight

	return b
}

func main() {
	servers := []*wrrServer{
		{"zoker.server.wtm:5555", 5, 0},
		{"zoker.server.wtm:2222", 2, 0},
		{"zoker.server.wtm:1111", 1, 0},
	}

	for i := 0; i < 20; i++ {
		bestServer := getBestServer(servers)

		if bestServer == nil {
			continue
		}

		fmt.Printf("Selected server: %s Weight: %d\n", bestServer.address, bestServer.weight)
	}

}

执行结果:

Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:2222 Weight: 2
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:1111 Weight: 1
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:2222 Weight: 2
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:2222 Weight: 2
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:1111 Weight: 1
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:2222 Weight: 2
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:2222 Weight: 2
Selected server: zoker.server.wtm:5555 Weight: 5
Selected server: zoker.server.wtm:5555 Weight: 5

可以看到，是比较分散的，也就是我们称之为的「平滑」

实现基于 WRR 的反向代理

了解了加权轮询算法之后，我们来改造上面的分发代理，上面是直接取随机数然后取余来达到随机的效果的，我们来改造 generateReverseProxy()使它能够具备加权分发的能力

package main

import (
	"fmt"
	"net/http"
	"net/http/httputil"
	"net/url"
)

type Nodes struct {
	node          url.URL
	weight        int
	currentWeight int
}

// config backends and weight
var nodes = []*Nodes{
	{url.URL{Host: "127.0.0.1:7791", Scheme: "http"}, 5, 0},
	{url.URL{Host: "127.0.0.1:7792", Scheme: "http"}, 1, 0},
	{url.URL{Host: "127.0.0.1:7793", Scheme: "http"}, 1, 0},
}

// using wrr to select nodes
func getBestNode() (bestNode *Nodes) {
	allWeight := 0
	for _, node := range nodes {
		allWeight += node.weight
		node.currentWeight += node.weight

		if bestNode == nil || node.currentWeight > bestNode.currentWeight {
			bestNode = node
		}
	}

	bestNode.currentWeight -= allWeight
	fmt.Printf("Select %s %d %d \n", bestNode.node.Host, bestNode.weight, bestNode.currentWeight)
	return bestNode
}

func generateReverseProxy() *httputil.ReverseProxy {
	// generate director
	director := func(req *http.Request) {
		backend := getBestNode().node
		req.URL.Scheme = backend.Scheme
		req.URL.Host = backend.Host
		//fmt.Printf("Scheme: %s Host: %s Path: %s\n", req.Scheme, req.Host, req.RequestURI)
	}
	return &httputil.ReverseProxy{Director: director}
}

func main() {
	// generate a reverse proxy
	reverseProxy := generateReverseProxy()
	fmt.Println("Server started on port 7788...")
	if err := http.ListenAndServe("127.0.0.1:7788", reverseProxy); err != nil {
		fmt.Printf("Server failed to start, error: %s \n", err.Error())
	}
}

启动服务后，我们模拟7次请求

for i in {1..7}; do curl 127.0.0.1:7788; done

输入图片说明

可以看到7次请求刚好按照权重分布，并且整个过程是按照 WRR 算法均匀分布的。

至此，一个基于加权轮询算法的反向代理就实现了，整个例子还缺少一些错误判定之类的工作，加上之后基本就可以使用了。

进阶思考

上面的例子我们通过修改请求的Scheme和Host来实现分发

		req.URL.Scheme = backend.Scheme
		req.URL.Host = backend.Host

我们打开url.go查看 URL 结构体的定义

// A URL represents a parsed URL (technically, a URI reference).
//
// The general form represented is:
//
//	[scheme:][//[userinfo@]host][/]path[?query][#fragment]
//
// URLs that do not start with a slash after the scheme are interpreted as:
//
//	scheme:opaque[?query][#fragment]
//
// Note that the Path field is stored in decoded form: /%47%6f%2f becomes /Go/.
// A consequence is that it is impossible to tell which slashes in the Path were
// slashes in the raw URL and which were %2f. This distinction is rarely important,
// but when it is, the code should use RawPath, an optional field which only gets
// set if the default encoding is different from Path.
//
// URL's String method uses the EscapedPath method to obtain the path. See the
// EscapedPath method for more details.
type URL struct {
	Scheme     string
	Opaque     string    // encoded opaque data
	User       *Userinfo // username and password information
	Host       string    // host or host:port
	Path       string    // path (relative paths may omit leading slash)
	RawPath    string    // encoded path hint (see EscapedPath method)
	ForceQuery bool      // append a query ('?') even if RawQuery is empty
	RawQuery   string    // encoded query values, without '?'
	Fragment   string    // fragment for references, without '#'
}

有这么多属性可以使用，能够带给我们更多的思考：

  1. 通过请求的 Path 进行逻辑判定，可以指定后端服务器或者进行 URL rewrite （Gitee 就是使用 Path 进行路由的指定和分片的）
  2. 可以追加 URL RawQuery，比如添加一个from=xxx，告诉后端，这个请求是从哪个反代发出的
  3. 根据 User 以及 Path 实现鉴权，可以实现类似网关的功能
  4. 通过请求的 Host 和 Path 来判定转发到什么目录，可以实现类似于 Gitee Pages 的多租户静态网站功能
  5. …

（END）

Written by
Blog Logo

Zoker

Published

Supported by
Image

Zoker's Blog

专注、沉淀、提升

Back to Overview