Golang分布式注册中心实现流程讲解

Golang分布式注册中心 Golang注册中心 2023-05-19 17:05:38 124人浏览八月长安

摘要

目录动手实现一个分布式注册中心日志服务log/Server.Golog/Client.go主启动程序LogService服务启动与注册service/service.go服务注册与发

动手实现一个分布式注册中心

以一个日志微服务为例，将日志服务注册到注册中心展开！

日志服务

log/Server.go

其实这一个日志类的功能就是有基本的写文件功能，然后就是注册一个Http的接口去写日志进去

package log
import (
	"io/ioutil"
	stlog "log"
	"net/http"
	"os"
)
var log *stlog.Logger
type fileLog string
// 编写日志的方法
func (fl fileLog) Write(data []byte) (int, error) {
	f, err := os.OpenFile(string(fl), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600)
	if err != nil {
		return 0, err
	}
	defer f.Close()
	return f.Write(data)
}
// 启动一个日志对象 参数为日志文件名
func Run(destination string) {
	log = stlog.New(fileLog(destination), "[go] - ", stlog.LstdFlags)
}
// 自身注册的一个服务方法
func RegisterHandlers() {
	http.HandleFunc("/log", func(w http.ResponseWriter, r *http.Request) {
		switch r.Method {
		case http.MethodPost:
			msg, err := ioutil.ReadAll(r.Body)
			if err != nil || len(msg) == 0 {
				w.WriteHeader(http.StatusBadRequest)
				return
			}
			write(string(msg))
		default:
			w.WriteHeader(http.StatusMethodNotAllowed)
			return
		}
	})
}
func write(message string) {
	log.Printf("%v\n", message)
}

log/Client.go

提供给外部服务的接口，定义好日志的命名格式，来显示调用接口去使用已经注册好的日志接口并且返回状态

package log
import (
	"bytes"
	"distributed/registry"
	"fmt"
	"net/http"
	stlog "log"
)
func SetClientLogger(serviceURL string, clientService registry.ServiceName) {
	stlog.SetPrefix(fmt.Sprintf("[%v] - ", clientService))
	stlog.SetFlags(0)
	stlog.SetOutput(&clientLogger{url: serviceURL})
}
type clientLogger struct {
	url string
}
func (cl clientLogger) Write(data []byte) (int, error) {
	b := bytes.NewBuffer([]byte(data))
	res, err := http.Post(cl.url+"/log", "text/plain", b)
	if err != nil {
		return 0, err
	}
	if res.StatusCode != http.StatusOK {
		return 0, fmt.Errorf("Failed to send log message. Service responded with %d - %s", res.StatusCode, res.Status)
	}
	return len(data), nil
}

主启动程序LogService

启动服务Logservice,主要执行start方法，里面有细节实现服务注册与服务发现

package main
import (
	"context"
	"distributed/log"
	"distributed/registry"
	"distributed/service"
	"fmt"
	stlog "log"
)
func main() {
  // 初始化启动一个日志文件对象
	log.Run("./distributed.log")
  // 日志服务注册的端口和地址
	host, port := "localhost", "4000"
	serviceAddress := fmt.Sprintf("http://%s:%s", host, port)
  // 初始化注册对象
	r := registry.Registration{
		ServiceName:      registry.LogService, // 自身服务名
		ServiceURL:       serviceAddress,  // 自身服务地址
		RequiredServices: make([]registry.ServiceName, 0),// 依赖服务
		ServiceUpdateURL: serviceAddress + "/services", // 服务列表
		HeartbeatURL: serviceAddress + "/heartbeat",  // 心跳
	}
  // 启动日志服务包含服务注册,发现等细节
	ctx, err := service.Start(
		context.Background(),
		host,
		port,
		r,
		log.RegisterHandlers,
	)
  // 异常写入到日志中
	if err != nil {
		stlog.Fatalln(err)
	}
  // 超时停止退出服务
	<-ctx.Done()
	fmt.Println("Shutting down log service.")
}

服务启动与注册

service/service.go

Start 启动服务的主方法


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
}

startService

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
	// 该协程为监听http服务，并且停止服务的时候cancel
	go func() {
		log.Println(srv.ListenAndServe())
		// 删除对应的服务
		err := registry.ShutdownService(fmt.Sprintf("http://%s:%s", host, port))
		if err != nil {
			log.Println(err)
		}
		cancel()
	}()
	// 该协程为监听手动停止服务的信号
	go func() {
		fmt.Printf("%v started. Press any key to stop. \n", serviceName)
		var s string
		fmt.Scanln(&s)
		err := registry.ShutdownService(fmt.Sprintf("http://%s:%s", host, port))
		if err != nil {
			log.Println(err)
		}
		srv.Shutdown(ctx)
		cancel()
	}()
	return ctx
}

服务注册与发现

registry/client.go

注册服务的时候会连着心跳以及服务更新的方法一起注册！

而服务更新里面的细节就是自己自定义了一个Handler然后ServeHttp方法里面去update全局的服务提供对象,

update主要是更新服务和删除服务的最新消息

然后就是提供一个注销服务的方法

package registry
import (
	"bytes"
	"encoding/JSON"
	"fmt"
	"log"
	"math/rand"
	"net/http"
	"net/url"
	"sync"
)
// 注册服务
func RegisterService(r Registration) error {
	// 获得心跳地址并注册
	heartbeatURL, err := url.Parse(r.HeartbeatURL)
	if err != nil {
		return err
	}
	http.HandleFunc(heartbeatURL.Path, func (w http.ResponseWriter, r *http.Request)  {
		w.WriteHeader(http.StatusOK)
	})
   // 获得服务更新地址，并且自定义http服务的handler，因为每次更新服务的时候，可以在ServeHttp方法里面去维护
	serviceUpdateURL, err := url.Parse(r.ServiceUpdateURL)
	if err != nil {
		return err
	}
	http.Handle(serviceUpdateURL.Path, &serviceUpdateHanlder{})
    // 写入buf值将服务对象发送给注册中心的services地址
	buf := new(bytes.Buffer)
	enc := json.NewEncoder(buf)
	err = enc.Encode(r)
	if err != nil {
		return err
	}
	res, err := http.Post(ServicesURL, "application/json", buf)
	if err != nil {
		return err
	}
	if res.StatusCode != http.StatusOK {
		return fmt.Errorf("Failed to register service. Registry service "+
			"responded with code %v", res.StatusCode)
	}
	return nil
}
type serviceUpdateHanlder struct{}
func (suh serviceUpdateHanlder) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		w.WriteHeader(http.StatusMethodNotAllowed)
		return
	}
	dec := json.NewDecoder(r.Body)
	var p patch
	err := dec.Decode(&p)
	if err != nil {
		log.Println(err)
		w.WriteHeader(http.StatusBadRequest)
		return
	}
	fmt.Printf("Updated received %v\n", p)
	prov.Update(p) // 更新服务提供对象
}
// 删除对应注册中心的服务地址
func ShutdownService(url string) error {
	req, err := http.NewRequest(http.MethodDelete, ServicesURL,
		bytes.NewBuffer([]byte(url)))
	if err != nil {
		return err
	}
	req.Header.Add("Content-Type", "text/plain")
	res, err := http.DefaultClient.Do(req)
	if err != nil {
		return err
	}
	if res.StatusCode != http.StatusOK {
		return fmt.Errorf("Failed to deregister service. Registry "+
			"service responded with code %v", res.StatusCode)
	}
	return nil
}
// 更新服务列表
func (p *providers) Update(pat patch) {
	p.mutex.Lock()
	defer p.mutex.Unlock()
    // 将patch中有新增的进行添加
	for _, patchEntry := range pat.Added {
		if _, ok := p.services[patchEntry.Name]; !ok {
			p.services[patchEntry.Name] = make([]string, 0)
		}
		p.services[patchEntry.Name] = append(p.services[patchEntry.Name],
			patchEntry.URL)
	}
    // 将patch中被标记删除的
	for _, patchEntry := range pat.Removed {
		if providerURLs, ok := p.services[patchEntry.Name]; ok {
			for i := range providerURLs {
				if providerURLs[i] == patchEntry.URL {
					p.services[patchEntry.Name] = append(providerURLs[:i],
						providerURLs[i+1:]...)
				}
			}
		}
	}
}
// 根据服务名负载均衡随机获取服务地址
func (p providers) get(name ServiceName) (string, error) {
	providers, ok := p.services[name]
	if !ok {
		return "", fmt.Errorf("No providers available for service %v", name)
	}
	idx := int(rand.Float32() * float32(len(providers)))
	return providers[idx], nil
}
// 对外暴露生产者的方法
func GetProvider(name ServiceName) (string, error) {
	return prov.get(name)
}
type providers struct {
	services map[ServiceName][]string
	mutex    *sync.RWMutex
}
// 服务提供对象
var prov = providers{
	services: make(map[ServiceName][]string),  // 服务列表  服务名->集群地址集合
	mutex:    new(sync.RWMutex),  // 锁  防止服务注册更新时的并发情况
}

registry/registration.go

主要是一些关于服务使用到的参数以及对象!

package registry
type Registration struct {
	ServiceName      ServiceName   // 服务名
	ServiceURL       string        // 服务地址
	RequiredServices []ServiceName // 依赖的服务
	ServiceUpdateURL string        // 服务更新的地址
	HeartbeatURL     string        // 心跳地址
}
type ServiceName string
// 服务名集合
const (
	LogService     = ServiceName("LogService")
	GradingService = ServiceName("GradingService")
	PortalService  = ServiceName("Portald")
)
// 服务对象参数
type patchEntry struct {
	Name ServiceName
	URL  string
}
// 更新的服务对象参数  
type patch struct {
	Added   []patchEntry
	Removed []patchEntry
}

registry/server.go

服务端的注册中心服务的增删改查管理以及心跳检测，及时将最新的更新的服务消息通知回给客户端

package registry
import (
	"bytes"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"sync"
	"time"
)
const ServerPort = ":3000"
const ServicesURL = "http://localhost" + ServerPort + "/services"   // 注册中心地址
// 服务对象集合
type registry struct {
	registrations []Registration
	mutex         *sync.RWMutex
}
// 添加服务
func (r *registry) add(reg Registration) error {
	r.mutex.Lock()
	r.registrations = append(r.registrations, reg)
	r.mutex.Unlock()
	err := r.sendRequiredServices(reg)
	r.notify(patch{
		Added: []patchEntry{
			patchEntry{
				Name: reg.ServiceName,
				URL:  reg.ServiceURL,
			},
		},
	})
	return err
}
// 通知服务接口请求去刷新改变后到服务
func (r registry) notify(fullPatch patch) {
	r.mutex.RLock()
	defer r.mutex.RUnlock()
	for _, reg := range r.registrations {
		go func(reg Registration) {
			for _, reqService := range reg.RequiredServices {
				p := patch{Added: []patchEntry{}, Removed: []patchEntry{}}
				sendUpdate := false
				for _, added := range fullPatch.Added {
					if added.Name == reqService {
						p.Added = append(p.Added, added)
						sendUpdate = true
					}
				}
				for _, removed := range fullPatch.Removed {
					if removed.Name == reqService {
						p.Removed = append(p.Removed, removed)
						sendUpdate = true
					}
				}
				if sendUpdate {
					err := r.sendPatch(p, reg.ServiceUpdateURL)
					if err != nil {
						log.Println(err)
						return
					}
				}
			}
		}(reg)
	}
}
// 更新每个服务的依赖服务
func (r registry) sendRequiredServices(reg Registration) error {
	r.mutex.RLock()
	defer r.mutex.RUnlock()
	var p patch
	for _, serviceReg := range r.registrations {
		for _, reqService := range reg.RequiredServices {
			if serviceReg.ServiceName == reqService {
				p.Added = append(p.Added, patchEntry{
					Name: serviceReg.ServiceName,
					URL:  serviceReg.ServiceURL,
				})
			}
		}
	}
	err := r.sendPatch(p, reg.ServiceUpdateURL)
	if err != nil {
		return err
	}
	return nil
}
// 告诉客户端更新，最新的服务列表是这个
func (r registry) sendPatch(p patch, url string) error {
	d, err := json.Marshal(p)
	if err != nil {
		return err
	}
	_, err = http.Post(url, "application/json", bytes.NewBuffer(d))
	if err != nil {
		return err
	}
	return nil
}
// 注册中心删除服务对象
func (r *registry) remove(url string) error {
	for i := range reg.registrations {
		if reg.registrations[i].ServiceURL == url {
			// 通知客户端更新对象信息
			r.notify(patch{
				Removed: []patchEntry{
					{
						Name: r.registrations[i].ServiceName,
						URL:  r.registrations[i].ServiceURL,
					},
				},
			})
			r.mutex.Lock()
			reg.registrations = append(reg.registrations[:i], reg.registrations[i+1:]...)
			r.mutex.Unlock()
			return nil
		}
	}
	return fmt.Errorf("Service at URL %s not found", url)
}
// 心跳检测
func (r *registry) heartbeat(freq time.Duration) {
	for {
		var wg sync.WaitGroup
		for _, reg := range r.registrations {
			wg.Add(1)
			go func(reg Registration) {
				defer wg.Done()
				success := true
				for attemps := 0; attemps < 3; attemps++ {
					res, err := http.Get(reg.HeartbeatURL)
					if err != nil {
						log.Println(err)
					} else if res.StatusCode == http.StatusOK {
						log.Printf("Heartbeat check passed for %v", reg.ServiceName)
						// 如果心跳恢复了，把服务重新注册回来
						if !success {
							r.add(reg)
						}
						break;
					}
					// 如果执行到这就代表着心跳没有响应，那就代表着需要回收注销该服务了
					log.Printf("Heartbeat check failed for %v", reg.ServiceName)
					if success {
						success = false
						r.remove(reg.ServiceURL)
					}
					time.Sleep(1 * time.Second)
				}
			}(reg)
			wg.Wait()
			time.Sleep(freq)
		}
	}
}
var once sync.Once
func SetupRegistryService() {
	// 保证执行一次进行服务到心跳 每三秒循环一遍
	once.Do(func() {
		go reg.heartbeat(3 * time.Second)
	})
}
var reg = registry{
	registrations: make([]Registration, 0),
	mutex:         new(sync.RWMutex),
}
type RegistryService struct{}
func (s RegistryService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	log.Println("Request received")
	switch r.Method {
	case http.MethodPost:
		dec := json.NewDecoder(r.Body)
		var r Registration
		err := dec.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
	}
}

到此这篇关于golang分布式注册中心实现流程讲解的文章就介绍到这了,更多相关Golang分布式注册中心内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

您可能感兴趣的文档:

Go语言（Golang）教程

点击免费下载>>软考高级考试备考技巧/历年真题/备考精华资料

--结束END--

本文标题: Golang分布式注册中心实现流程讲解

本文链接: https://www.lsjlt.com/news/213525.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

本篇文章演示代码以及资料文档资料下载

下载Word文档到电脑，方便收藏和打印～

下载Word文档

去做题

猜你喜欢

Golang分布式注册中心实现流程讲解

目录动手实现一个分布式注册中心日志服务log/Server.golog/Client.go主启动程序LogService服务启动与注册service/service.go服务注册与发...

99+

2023-05-19

Golang分布式注册中心 Golang注册中心
SpringBoot实现服务接入nacos注册中心流程详解

目录概述接入nacos注册中心springboot服务pom文件application.properties配置源码分析小结源码流程图概述某些场景下只需要把springboot微服...

99+

2023-01-13

SpringBoot接入nacos注册中心 SpringBoot接入nacos
SpringCloud注册中心部署Eureka流程详解

目录1、Eureka服务2、服务提供者3、服务消费者4、服务调用测试今天我们开始正式编码，如何创建spring boot项目这篇文章就不再讲述，如果想要了解可以阅读我之前的创建spr...

99+

2022-11-16

SpringCloud注册中心部署Eureka SpringCloud Eureka SpringCloud注册中心
搭建JEESZ分布式架构7--Zookeeper注册中心安装

ZooKeeper是一个分布式开源框架，提供了协调分布式应用的基本服务，它向外部应用暴露一组通用服务——分布式同步（Distributed Synchronization）.命名服务（Naming Service）.集群维护（Group M...

99+

2023-06-03
Spring Cloud Eureka服务注册中心入门流程分析

目录项目搭建客户端注册聚合层处理Eureka架构集群搭建多区域配置自我保护开关心跳机制Eureka实例信息存储项目地址在学习Ribbon使用的时候，我们是直接在配置文件中写死服务地址...

99+

2024-04-02
SpringCloud通过Nacos实现注册中心与远程服务调用详解流程

目录1. 基于Nacos实现服务注册与发现1.1 pom依赖1.2 yaml配置1.3 添加启动注解1.4 启动服务查看控制台2.基于Nacos实现远程服务调用2.1 客户端创建Re...

99+

2024-04-02
详解Nacos中注册中心和配置中心的实现

目录1.Nacos 简介Nacos 特性介绍2.注册中心实现2.1 创建服务提供者2.2 创建服务消费者3.配置中心实现3.1 新建项目并添加依赖3.2 配置 Nacos Confi...

99+

2024-04-02
SpringBoot大学心理服务系统实现流程分步讲解

目录一、前言介绍二、系统功能分析三、系统详细设计四、部分核心代码一、前言介绍在引言中，将论述大学生心理服务系统的当前背景以及系统开发的目的，后续章节将严格按照软件开发流程，对系统进...

99+

2024-04-02
Redis实现分布式锁的实例讲解

在一个分布式系统中，会遇到一些需要对多个节点共享的资源加锁的情况，这个时候需要用到分布式锁。分布式锁通常保存在一个共享的存储系统中，可以被多个节点共享和访问。锁的本质简单来讲，锁...

99+

2024-04-02
Java Zookeeper分布式分片算法超详细讲解流程

目录背景技术方案分布式协调中间件基于Zookeeper的技术方案服务注册目录设计服务分片处理流程编码实现总结背景公司的一个服务需要做类似于分片的逻辑，一开始服务基于传统部署方式通过...

99+

2023-03-01

Java Zookeeper分布式分片算法 Java Zookeeper Java分布式分片算法
GoLang实现日志收集器流程讲解

目录Go自己动手实现日志收集1. 日志器的接口2. 命令写入日志器3. 文件写入日志器4. Demo演示5. 总结和提高Go自己动手实现日志收集看日志是开发者平时排查BUG所必须的...

99+

2023-05-20

GoLang日志收集 Go日志收集
分布式系统中使用 Golang 函数实现分布式锁

golang 函数可用于实现分布式锁，协调多个进程对共享资源的访问。这些函数通过利用共享存储（如 redis）来实现锁机制，确保在任何时刻只有一个进程能访问资源。基于 Golang ...

99+

2024-04-20

golang 分布式系统 redis git
Mybatis实现查询相册数据列表流程讲解

目录1.书写执行的SQL语句2.在项目的根包下创建pojo.vo.AlbumListItemVO类3.在AlbumMapper.java中添加抽象方法4.在AlbumMapper.x...

99+

2022-12-09

Mybatis查询相册数据列表 Mybatis查询数据
Golang协程在分布式系统的实现

问题：协程在分布式系统中的实现如何？答案：goroutine 创建：使用 go 关键字创建协程。通道通信：通过创建通道来安全地交换数据。实战案例：协程池用于分布式任务处理，提高性能。优势...

99+

2024-04-15

golang 协程同步机制
Zookeeper如何实现分布式服务配置中心详解

目录1 Linux安装并启动Zookeeper1.1 安装1.1.1 安装1.2 启动3 Spring Boot配置 3.1 依赖3.2 配置文件3.3 项目代码3.4 启动测试总结...

99+

2024-04-02
5分钟搭建SpringCloud Eureka服务注册中心的实现

创建父级项目只需保留pom.xml文件这里只需搭建一个微服务其他操作并无 <?xml version="1.0" encoding="UTF-8"?...

99+

2024-04-02
如何在Redis中实现分布式限流

在Redis中实现分布式限流可以通过使用Redis的计数器和过期时间来实现。以下是一种基于计数器和过期时间的简单实现方式：使用R...

99+

2024-04-09

Redis
SpringBean和Controller实现动态注册与注销过程详细讲解

目录说明注册和注销工具类编写测试用例测试结果注册Service注册controller注销Controller部分场景下可能需要下载远程jar包，然后注册jar包中的Bean和Con...

99+

2023-02-09

SpringBean动态注册与注销 SpringBean Controller注册与注销
Zookeeper如何实现分布式服务配置中心

本篇文章为大家展示了Zookeeper如何实现分布式服务配置中心，内容简明扼要并且容易理解，绝对能使你眼前一亮，通过这篇文章的详细介绍希望你能有所收获。1 Linux安装并启动Zookeeper1.1 安装下载链接：https://arch...

99+

2023-06-25
c# 理解csredis库实现分布式锁的详细流程

声明：这里首先使用的是csredis,地址是https://github.com/2881099/csredis 该库本身已经足够完善，这里我画蛇添足一下，为了方便自己的使用。本...

99+

2024-04-02