基于consul构建golang系统分布式服务发现机制

2024-04-02 04:48
文章标签 系统 golang 构建 发现 机制 分布式服务 consul

本文主要是介绍基于consul构建golang系统分布式服务发现机制，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

原文链接：石匠1号的Blog

在分布式架构中，服务治理是一个重要的问题。在没有服务治理的分布式集群中，各个服务之间通过手工或者配置的方式进行服务关系管理，遇到服务关系变化或者增加服务的时候，人肉配置极其麻烦且容易出错。

之前在一个C++项目中，采用ZooKeeper进行服务治理，可以很好的维护服务之间的关系，但是使用起来较为麻烦。现在越来越多新的项目采用consul进行服务治理，各方面的评价都优于ZooKeeper，经过几天的研究，这里做一个总结。

zookeeper和consul比较
  • 开发语言方面，zookeeper采用java开发，安装的时候需要部署java环境；consul采用golang开发，所有依赖都编译到了可执行程序中，即插即用。
  • 部署方面，zookeeper一般部署奇数个节点方便做简单多数的选举机制。consul部署的时候分server节点和client节点（通过不同的启动参数区分），server节点做leader选举和数据一致性维护，client节点部署在服务机器上，作为服务程序访问consul的接口。
  • 一致性协议方面，zookeeper使用自定义的zab协议，consul的一致性协议采用更流行的Raft。
  • zookeeper不支持多数据中心，consul可以跨机房支持多数据中心部署，有效避免了单数据中心故障不能访问的情况。
  • 链接方式上，zookeeper client api和服务器保持长连接，需要服务程序自行管理和维护链接有效性，服务程序注册回调函数处理zookeeper事件，并自己维护在zookeeper上建立的目录结构有效性（如临时节点维护）；consul 采用DNS或者http获取服务信息，没有主动通知，需要自己轮训获取。
  • 工具方面，zookeeper自带一个cli_mt工具，可以通过命令行登录zookeeper服务器，手动管理目录结构。consul自带一个Web UI管理系统， 可以通过参数启动并在浏览器中直接查看信息。
consul相关资源
  • 可执行程序下载地址： https://www.consul.io/downloads.html
  • 官方说明文档： https://www.consul.io/docs/index.html
  • api说明文档： https://godoc.org/github.com/hashicorp/consul/api#pkg-index
  • golang api代码位置：github.com/hashicorp/consul/api

linux系统中，下载consul可执行程序后直接拷贝到/usr/local/bin就可以使用了，无需其他额外配置。

服务节点启动方式：

consul agent -server -bootstrap-expect 1 -data-dir /tmp/consul -node=service-center -bind=192.168.0.2 -client 0.0.0.0 -ui -config-dir /etc/consul.d/

参数说明：

  • -server 表示以server节点模式启动consul
  • -bootstrap-expect 1 表示期待的server节点一共有几个，如3个server集群模式
  • -data-dir consul存储数据的目录
  • -node 节点的名字
  • -bind 绑定的服务ip
  • -client 0.0.0.0 -ui 启动Web UI管理工具
  • -config-dir 指定服务配置文件的目录（这个目录下的所有.json文件，作为服务配置文件读取）
consul服务发现机制测试

为了测试consul服务治理方式，设定如下场景：

一个manager类型的服务，需要根据负载来管理若干worker类型的服务并进行业务通信；而worker服务也需要知道manager提供的内部服务接口地址做业务交互。即manger和worker都需要互相知道对方的通信地址。

做如下规则设定准备：

  • manager和worker都需要向consul注册自己的服务，让对方发现自己的服务地址（ip和端口）
  • 采用consul的key-value存储机制，worker周期性更新自己的负载信息到相应的key；manger从worker的key中获取负载信息，并同步更新到本地。
  • 服务类型规则： manager的服务类型用字符串"manager"表示，各个worker的服务类型采用字符串"worker"表示。
  • 服务注册ID规则： 服务类型-服务IP，如 manager-192.168.0.2
  • key的构建规则: 服务类型/IP:Port, 如 worker/192.168.0.2:5400
  • 存储的数据采用json格式：{"load":100,"ts":1482828232}
golang测试程序
package mainimport ("encoding/json""flag""fmt""github.com/hashicorp/consul/api""log""math/rand""net/http""os""os/signal""strconv""strings""sync""time"
)type ServiceInfo struct {ServiceID stringIP        stringPort      intLoad      intTimestamp int //load updated ts
}
type ServiceList []ServiceInfotype KVData struct {Load      int `json:"load"`Timestamp int `json:"ts"`
}var (servics_map     = make(map[string]ServiceList)service_locker  = new(sync.Mutex)consul_client   *api.Clientmy_service_id   stringmy_service_name stringmy_kv_key       string
)func CheckErr(err error) {if err != nil {log.Printf("error: %v", err)os.Exit(1)}
}
func StatusHandler(w http.ResponseWriter, r *http.Request) {fmt.Println("check status.")fmt.Fprint(w, "status ok!")
}func StartService(addr string) {http.HandleFunc("/status", StatusHandler)fmt.Println("start listen...")err := http.ListenAndServe(addr, nil)CheckErr(err)
}func main() {var status_monitor_addr, service_name, service_ip, consul_addr, found_service stringvar service_port intflag.StringVar(&consul_addr, "consul_addr", "localhost:8500", "host:port of the service stuats monitor interface")flag.StringVar(&status_monitor_addr, "monitor_addr", "127.0.0.1:54321", "host:port of the service stuats monitor interface")flag.StringVar(&service_name, "service_name", "worker", "name of the service")flag.StringVar(&service_ip, "ip", "127.0.0.1", "service serve ip")flag.StringVar(&found_service, "found_service", "worker", "found the target service")flag.IntVar(&service_port, "port", 4300, "service serve port")flag.Parse()my_service_name = service_nameDoRegistService(consul_addr, status_monitor_addr, service_name, service_ip, service_port)go DoDiscover(consul_addr, found_service)go StartService(status_monitor_addr)go WaitToUnRegistService()go DoUpdateKeyValue(consul_addr, service_name, service_ip, service_port)select {}
}func DoRegistService(consul_addr string, monitor_addr string, service_name string, ip string, port int) {my_service_id = service_name + "-" + ipvar tags []stringservice := &api.AgentServiceRegistration{ID:      my_service_id,Name:    service_name,Port:    port,Address: ip,Tags:    tags,Check: &api.AgentServiceCheck{HTTP:     "http://" + monitor_addr + "/status",Interval: "5s",Timeout:  "1s",},}client, err := api.NewClient(api.DefaultConfig())if err != nil {log.Fatal(err)}consul_client = clientif err := consul_client.Agent().ServiceRegister(service); err != nil {log.Fatal(err)}log.Printf("Registered service %q in consul with tags %q", service_name, strings.Join(tags, ","))
}func WaitToUnRegistService() {quit := make(chan os.Signal, 1)signal.Notify(quit, os.Interrupt, os.Kill)<-quitif consul_client == nil {return}if err := consul_client.Agent().ServiceDeregister(my_service_id); err != nil {log.Fatal(err)}
}func DoDiscover(consul_addr string, found_service string) {t := time.NewTicker(time.Second * 5)for {select {case <-t.C:DiscoverServices(consul_addr, true, found_service)}}
}func DiscoverServices(addr string, healthyOnly bool, service_name string) {consulConf := api.DefaultConfig()consulConf.Address = addrclient, err := api.NewClient(consulConf)CheckErr(err)services, _, err := client.Catalog().Services(&api.QueryOptions{})CheckErr(err)fmt.Println("--do discover ---:", addr)var sers ServiceListfor name := range services {servicesData, _, err := client.Health().Service(name, "", healthyOnly,&api.QueryOptions{})CheckErr(err)for _, entry := range servicesData {if service_name != entry.Service.Service {continue}for _, health := range entry.Checks {if health.ServiceName != service_name {continue}fmt.Println("  health nodeid:", health.Node, " service_name:", health.ServiceName, " service_id:", health.ServiceID, " status:", health.Status, " ip:", entry.Service.Address, " port:", entry.Service.Port)var node ServiceInfonode.IP = entry.Service.Addressnode.Port = entry.Service.Portnode.ServiceID = health.ServiceID//get data from kv stores := GetKeyValue(service_name, node.IP, node.Port)if len(s) > 0 {var data KVDataerr = json.Unmarshal([]byte(s), &data)if err == nil {node.Load = data.Loadnode.Timestamp = data.Timestamp}}fmt.Println("service node updated ip:", node.IP, " port:", node.Port, " serviceid:", node.ServiceID, " load:", node.Load, " ts:", node.Timestamp)sers = append(sers, node)}}}service_locker.Lock()servics_map[service_name] = sersservice_locker.Unlock()
}func DoUpdateKeyValue(consul_addr string, service_name string, ip string, port int) {t := time.NewTicker(time.Second * 10)for {select {case <-t.C:StoreKeyValue(consul_addr, service_name, ip, port)}}
}func StoreKeyValue(consul_addr string, service_name string, ip string, port int) {my_kv_key = my_service_name + "/" + ip + ":" + strconv.Itoa(port)var data KVDatadata.Load = rand.Intn(100)data.Timestamp = int(time.Now().Unix())bys, _ := json.Marshal(&data)kv := &api.KVPair{Key:   my_kv_key,Flags: 0,Value: bys,}_, err := consul_client.KV().Put(kv, nil)CheckErr(err)fmt.Println(" store data key:", kv.Key, " value:", string(bys))
}func GetKeyValue(service_name string, ip string, port int) string {key := service_name + "/" + ip + ":" + strconv.Itoa(port)kv, _, err := consul_client.KV().Get(key, nil)if kv == nil {return ""}CheckErr(err)return string(kv.Value)
}

程序通过参数控制自己启动的服务角色类型和需要发现的服务类型。传入的consul_addr是本机consul client agent的地址，一般是loacalhost:8500 。 由于consul集成了服务健康检查，所以服务需要启动一个检查接口，这里启动一个http服务来做响应。

consul集群启动

启动3个consul server :

consul agent -server -bootstrap-expect 3 -data-dir /tmp/consul -node=server001 -bind=10.2.1.54
consul agent -server  -data-dir /tmp/consul -node=server002 -bind=10.2.1.83 -join 10.2.1.54
consul agent -server  -data-dir /tmp/consul -node=server003 -bind=10.2.1.80 -join 10.2.1.54

server001-003构成了一个3个server node的consul集群。先启动server001，并指定需要3个server node构成集群，server002和server003启动的时候指定加入（-join）server001.

启动一个manger：

consul agent -data-dir /tmp/consul -node=mangaer -bind=10.2.1.92  -join 10.2.1.54
./service -consul_addr=127.0.0.1:8500 -monitor_addr=127.0.0.1:54321 -service_name=manager -ip=10.2.1.92 -port=4300 -found_service=worker

启动2个worker:

consul agent -data-dir /tmp/consul -node=worker001 -bind=10.2.1.93  -join 10.2.1.54
./service -consul_addr=127.0.0.1:8500 -monitor_addr=127.0.0.1:54321 -service_name=worker -ip=10.2.1.93 -port=4300 -found_service=manager
consul agent -data-dir /tmp/consul -node=worker002 -bind=10.2.1.94  -join 10.2.1.54
./service -consul_addr=127.0.0.1:8500 -monitor_addr=127.0.0.1:54321 -service_name=worker -ip=10.2.1.94 -port=4300 -found_service=manager

service程序是前面部分代码编译后的测试程序。

这样就构建了3个server node的consul集群，以及1各manager和2个worker的分布式服务程序，他们可以互相发现对方，并且manager可以获取到worker的负载情况，实现了互通。

结束

通过使用consul的服务注册发现机制和key-value存储机制，实现了服务发现以及manager获取worker服务负载数据的机制。由于consul的发现机制不能进行更多的数据交互，所以只能使用key-value机制配合进行数据共享（zookeeper中数据可以存储在节点上）。如果业务有进一步需求，可以方便的扩展存储的数据结构来实现。

以上的测试程序既有服务注册，存储数据更新，也有服务发现和数据获取，但是代码量比zookeeper机制少很多，因为zookeeper需要自己建立和维护目录树，注册和处理zookeeper event事件，监控zookeeper的链接并处理重连和信息重建等健康管理工作。

总的来说，consul比zookeeper使用简单易用很多。可以在新项目中尝试使用，特别是golang项目，技术栈也比较统一。

这篇关于基于consul构建golang系统分布式服务发现机制的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/869004。

相关文章

Spring Security 单点登录与自动登录机制的实现原理

Spring Security 单点登录与自动登录机制的实现原理

《SpringSecurity单点登录与自动登录机制的实现原理》本文探讨SpringSecurity实现单点登录（SSO）与自动登录机制,涵盖JWT跨系统认证、RememberMe持久化Token... 目录一、核心概念解析1.1 单点登录（SSO）1.2 自动登录（Remember Me）二、代码分析三、
阅读更多...
Linux系统中查询JDK安装目录的几种常用方法

Linux系统中查询JDK安装目录的几种常用方法

《Linux系统中查询JDK安装目录的几种常用方法》：本文主要介绍Linux系统中查询JDK安装目录的几种常用方法,方法分别是通过update-alternatives、Java命令、环境变量及目... 目录方法 1：通过update-alternatives查询（推荐）方法 2：检查所有已安装的 JDK方
阅读更多...
Linux系统之lvcreate命令使用解读

Linux系统之lvcreate命令使用解读

《Linux系统之lvcreate命令使用解读》lvcreate是LVM中创建逻辑卷的核心命令,支持线性、条带化、RAID、镜像、快照、瘦池和缓存池等多种类型,实现灵活存储资源管理,需注意空间分配、R... 目录lvcreate命令详解一、命令概述二、语法格式三、核心功能四、选项详解五、使用示例1. 创建逻
阅读更多...
Go语言并发之通知退出机制的实现

Go语言并发之通知退出机制的实现

《Go语言并发之通知退出机制的实现》本文主要介绍了Go语言并发之通知退出机制的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录1、通知退出机制1.1 进程/main函数退出1.2 通过channel退出1.3 通过cont
阅读更多...
Spring Boot 中的默认异常处理机制及执行流程

Spring Boot 中的默认异常处理机制及执行流程

《SpringBoot中的默认异常处理机制及执行流程》SpringBoot内置BasicErrorController,自动处理异常并生成HTML/JSON响应,支持自定义错误路径、配置及扩展,如... 目录Spring Boot 异常处理机制详解默认错误页面功能自动异常转换机制错误属性配置选项默认错误处理
阅读更多...
Java中的xxl-job调度器线程池工作机制

Java中的xxl-job调度器线程池工作机制

《Java中的xxl-job调度器线程池工作机制》xxl-job通过快慢线程池分离短时与长时任务,动态降级超时任务至慢池,结合异步触发和资源隔离机制,提升高频调度的性能与稳定性,支撑高并发场景下的可靠... 目录⚙️ 一、调度器线程池的核心设计 二、线程池的工作流程 三、线程池配置参数与优化 四、总结：线程
阅读更多...
Spring Boot Maven 插件如何构建可执行 JAR 的核心配置

Spring Boot Maven 插件如何构建可执行 JAR 的核心配置

《SpringBootMaven插件如何构建可执行JAR的核心配置》SpringBoot核心Maven插件,用于生成可执行JAR/WAR,内置服务器简化部署,支持热部署、多环境配置及依赖管理... 目录前言一、插件的核心功能与目标1.1 插件的定位1.2 插件的 Goals（目标）1.3 插件定位1.4 核
阅读更多...
使用Python构建一个高效的日志处理系统

使用Python构建一个高效的日志处理系统

《使用Python构建一个高效的日志处理系统》这篇文章主要为大家详细讲解了如何使用Python开发一个专业的日志分析工具,能够自动化处理、分析和可视化各类日志文件,大幅提升运维效率,需要的可以了解下... 目录环境准备工具功能概述完整代码实现代码深度解析1. 类设计与初始化2. 日志解析核心逻辑3. 文件处
阅读更多...
golang程序打包成脚本部署到Linux系统方式

golang程序打包成脚本部署到Linux系统方式

《golang程序打包成脚本部署到Linux系统方式》Golang程序通过本地编译（设置GOOS为linux生成无后缀二进制文件）,上传至Linux服务器后赋权执行,使用nohup命令实现后台运行,完... 目录本地编译golang程序上传Golang二进制文件到linux服务器总结本地编译Golang程序
阅读更多...
Linux系统性能检测命令详解

Linux系统性能检测命令详解

《Linux系统性能检测命令详解》本文介绍了Linux系统常用的监控命令（如top、vmstat、iostat、htop等）及其参数功能,涵盖进程状态、内存使用、磁盘I/O、系统负载等多维度资源监控,... 目录toppsuptimevmstatIOStatiotopslabtophtopdstatnmon
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2