A distributed system with Go

从上学期开始，学校的教务系统更新了。新版的正方多了一个负载均衡的功能，但无奈实在做的太差了！为什么这么说尼？
因为在上学期抢课的时候入口网关依旧拉闸，而真正的后端服务器访问速度比德芙还丝滑。因此暂将锅甩给负载均衡服务器。既然学校的负载均衡拉闸，那么就有了 “帮学校做负载均衡” 的想法。
大体的思路就是，通过扫描内网网段发现隐藏的地址，然后负载均衡这些隐藏的地址。
既然是负载均衡，第一时间想到的就是Nginx了，通过资料搜集，大致确定了基本的流程，因此有了这么的这个流程图:

需要解决的问题

由于非官方的代理，因此我们无法知道教务系统部署在内网的哪些机器上，而校园网一般使用B类网，因此需要扫描大量的ip地址。
但这样无疑会触发学校网管的报警，正常情况下会导致ip短时间内被封。导致扫描质量低（由于被禁网，导致程序误认为扫描超时从而导致目标未被发现）。
基于这个问题，随之而生的想法就是分布式扫描，将扫描的工作打散到用户中，服务器只负责扫描几个主干网段。
一句话概括就是: 将扫描工作分散到多台机器上，流程图如下：

Todo

前情提示: 本文通信均采用HTTP、代码部分存在伪代码
代码仓库: (暂不开源)

1. 注册中心 -> 服务注册与服务发现
2. 日志服务 -> 记录日志
3. 数据库服务 -> 数据库操作(CRUD)
4. 扫描器服务 -> 发现教务系统地址
5. 测试器服务 -> 测试数据库中地址的健康度
6. Api服务 -> 向consul发送负载信息

大致结构如下：

注册中心

作为一个分布式系统，不可避免地需要服务注册与服务发现。因此需要一个注册中心来处理各个服务之间的依赖关系，在服务上线后通知依赖这个服务的服务(这里有点绕)
举个例子:
数据库的每个操作都需要记录日志，日志为了统一管理所以有一个日志服务专门处理日志信息。此时，日志服务因某些原因(可能是人为、也可能是网络掉线等)在数据库服务注册之后才注册，这是注册中心就需要通知数据库，让数据库的日志记录转成使用日志服务。

功能与实现

作为一个注册中心，首先我们需要一个web服务来接收服务发送的信息(注册、依赖更新、注销等等)，但在这之前我们先来定义一下我们要用到的结构(在面向对象中为类)

1. registry来表示注册操作
   • registrations来存放注册的服务
   • add 注册服务
   • notify 事件通知
   • sendRequiredServices 发送依赖的服务
   • sendPatch 发送依赖项
   • remove 移除(注销)服务
   • Heartbeat 心跳包
2. Registration表示服务注册结构体
   • ServiceName 服务名
   • ServiceURL 服务地址
   • RequiredServices[数组]服务依赖项
   • ServiceUpdateURL 服务与注册中心沟通的URL
   • HeartbeatURL 心跳检测地址

type registry struct {
	registrations []Registration
	mutex         *sync.RWMutex
}
// 服务注册结构体
type Registration struct {
	ServiceName ServiceName
	ServiceURL  string
	RequiredServices []ServiceName
	ServiceUpdateURL string
	HeartbeatURL     string
}
type ServiceName string
// 添加注册
func (r *registry) add(reg Registration) error

// 当依赖服务运行时，通知依赖者
func (r registry) notify(fullpatch patch) 
// 注册中心向服务发送依赖相关内容
func (r registry) sendRequiredServices(reg Registration) error
// 发送依赖项
func (r registry) sendPatch(p patch, url string) error
// 移除注册（注销）
func (r *registry) remove(url string) error
// 心跳包
func (r *registry) Heartbeat(freq time.Duration) 

既然注册中心是作为一个web服务实现的，那么肯定是需要一个web server的，由于项目属于玩票性质，也不算大因此使用Go内置的net/http来实现，代码如下：

type RegistryService struct{}
func (s RegistryService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	log.Printf("Request received")
	switch r.Method {
	case http.MethodPost:
		decoder := json.NewDecoder(r.Body)
		var r Registration
		err := decoder.Decode(&r)
		if err != nil {
			log.Println(err)
			w.WriteHeader(http.StatusBadRequest)
			return
		}
		log.Printf("Adding service: %v with URL: %s\n", r.ServiceName, r.ServiceURL)
		err = reg.add(r)
		if err != nil {
			log.Println(err)
			w.WriteHeader(http.StatusBadRequest)
			return
		}
	case http.MethodDelete:
		payload, err := ioutil.ReadAll(r.Body)
		if err != nil {
			log.Println(err)
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		url := string(payload)
		log.Printf("Removing service at URL: %s", url)
		err = reg.remove(url)
		if err != nil {
			log.Println(err)
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
	default:
		w.WriteHeader(http.StatusMethodNotAllowed)
		return
	}
}

服务组件

对服务来说，就是根据注册中心定好的规则来注册服务，然后根据自身的依赖来处理对应的功能。
因为要处理相应的依赖，因此除了Registration外，再定义一个处理服务依赖的结构体及方法: providers.

type providers struct {
	services map[ServiceName][]string   // 服务名->服务url
	mutex    *sync.RWMutex
}
// 更新依赖
func (p *providers) Update(pat patch)
// // 通过服务名称，找到依赖的urls，从依赖项里面随机返回一个url
func (p providers) get(name ServiceName) (string, error)
// 导出给外部使用
func GetProvide(name ServiceName) (string, error)

组件的web服务

因为每个服务都需要使用注册这些通用的功能，且这部分的工作都是重复的，因此将web抽出来公用.

func Start(ctx context.Context, host, port string,
	reg registry.Registration,
	RegisterHandlersFunc func()) (context.Context, error) {

	RegisterHandlersFunc()
	ctx = startService(ctx, reg.ServiceName, host, port)
	err := registry.RegisterService(reg)
	if err != nil {
		return ctx, err
	}
	return ctx, nil
}

func startService(ctx context.Context, ServiceName registry.ServiceName,
	host, port string) context.Context {
	ctx, cancel := context.WithCancel(ctx)
	var srv http.Server
	srv.Addr = ":" + port
	address := fmt.Sprintf("http://%s:%s", host, port)

	go func() {
		log.Println(srv.ListenAndServe())
		// 关闭的时候要取消注册
		// ... todo
		err := registry.ShutdownService(address)
		if err != nil {
			log.Println(err)
		}
		cancel()
	}()

	go func() {
		fmt.Printf("Service is running in %v\n", address)
		fmt.Println("Registry service started. Press any key to stop.")
		var s string
		fmt.Scanln(&s)
		srv.Shutdown(ctx)
		err := registry.ShutdownService(address)
		if err != nil {
			log.Println(err)
		}
		cancel()
	}()
	return ctx
}

至此，可以开始专心的写业务了

扫描器

得益于Go高并发的优势，开启数百万个的goroutine的开销也不会很大，非常的轻量！🛫
因此代码实现起来很轻松，大体思路和端口扫描器类似，在此基础上根据目标特征添加判断条件即可
端口扫描器代码如下：

...
func worker(ports chan int, results chan int) {
	for p := range ports {
		address := fmt.Sprintf("192.168.2.122:%d", p)
		fmt.Printf("start %s\n", address)
		conn, err := net.DialTimeout("tcp", address, time.Second*5)
		if err != nil {
			results <- 0
			continue
		}
		conn.Close()
		results <- p
	}
}

func main() {
	ports := make(chan int, 30000)
	results := make(chan int, 100)
	scanEndPort := 65535
	var openPorts []int

	for i := 0; i < cap(ports); i++ {
		go worker(ports, results)
	}
	// push port to channel
	go func() {
		for i := 1; i <= scanEndPort; i++ {
			ports <- i
		}
		close(ports)
	}()

	for i := 1; i <= scanEndPort; i++ {
		res := <-results
		if res != 0 {
			openPorts = append(openPorts, res)
		}
	}
	close(results)
	sort.Ints(openPorts)
	for _, port := range openPorts {
		fmt.Printf("%d opend\n", port)
	}
}

测试器

测试器主要功能是从数据库中取出教务系统地址，然后测试。与扫描器不同仅在于扫描器是写，测试器是读。因此这部分内容和扫描器实际上是在同个包内的，只是逻辑上将它分离了出来。

这部分其实和consul的功能是重复的，因此代码不过多赘述

数据库

由于校园网中，教务系统的地址不会太多，因此数据库的选择十分的随意(不存在性能方面的要求),所以这里使用自己熟悉的redis作为数据库。
由于本系统是和官方的负载均衡并行的 因此存在某些结点用于两者共同访问导致压力上涨，响应不及时，因此利用redis的sorted-set在测试的时候，将响应快的地址设置高分数，使用sorted-sets的好处还有一个就是，集合的元素都是不重复的！，对于代理池来说，这个分数代表着教务系统地址稳定性的重要标准，因此设置分数的规则如下：

• 基本分：10分，最高分：20
• 测试器设置一个更严格的超时时间来判断结点是否流畅
• 分数基本分为可用，检测器会定时循环检测每个代理可用情况，一旦检测到有可用的代理则立即置为最高分
• 新获取的代理分数设置为基本分，如果测试可行(流畅)则置为满分，不可行(超时)则分数减一
• 分数减到0后代理移除

由于是内网环境，分数与超时时间可以根据实际情况设置更严格

主要功能就是简单的CRUD啦，本文只讲逻辑与伪代码，实现部分就不多说了。

// set jwglxt to max score
func max(addr discover.Addr) (err error) {
	member, err := addr.MarshalBinary()
	if err != nil {
		return err
	}
	err = rdb.ZAdd(ctx, redisConfig.Key, &redis.Z{
		Score:  SCORE_MAX,
		Member: member,
	}).Err()
	if err != nil {
		return err
	}
	return nil
}
// decrease proxy score
func decrease(addr discover.Addr) (err error) {
	member, err := addr.MarshalBinary()
	if err != nil {
		return err
	}
	err = rdb.ZIncrBy(ctx, redisConfig.Key, -1, string(member)).Err()
	if err != nil {
		return err
	}
	score := rdb.ZScore(ctx, redisConfig.Key, string(member))
	if score.Val() <= 0.00 {
		log.Printf("%v current score %v, remove.\n", addr, score.Val())
		err := rdb.ZRem(ctx, redisConfig.Key, member).Err()
		if err != nil {
			return err
		}
	}
	return nil
}
// add new proxy to redis
func add(addrs discover.Addrs) (err error) {
	for _, addr := range addrs {
		buf, err := addr.MarshalBinary()
		if err != nil {
			return nil
		}
		err = rdb.ZAdd(ctx, redisConfig.Key, &redis.Z{
			Score:  SCORE_DEFAULT,
			Member: buf,
		}).Err()
		if err != nil {
			return err
		}
	}
	return nil
}

consul-upsync-nginx实现动态负载均衡

该部分可以说是系统实现的关键了，因为nginx自带的负载均衡是写死的，不能根据后端情况动态调整，通过一番搜索对比，最终决定了consul + upsync方案。

1. upsync 一个Nginx的模块(扩展)
2. consul 一个分布式高可用的系统

这部分仅限于“能用”阶段，笔者也不太懂，就不乱说啦！
主要就是Api服务定时的获取数据库内容(分数作为权重)，然后推送到consul中

总结

该项目是学习Go时的一个练手项目，很多地方都不太好，因此仓库就不开源了🐕