Go语言优雅关闭与重启

原创

码农小辉

发布于 2022-09-07 10:16:27

1.6K0

发布于 2022-09-07 10:16:27

文章被收录于专栏：用户1381554的专栏

go优雅关闭与重启

背景

后端服务程序在配置更新，程序修改后发布的过程中存在一些未处理完成的请求，和当前服务中为落地的资源（缓存、记录、日志等数据），为了减少这种情况带来的数据异常，需要有一种机制，在服务收到重启或者关闭信号的同时进行一些数据收尾处理。

原理

处理服务优雅关闭和重启需要从下面几个方向完善服务的重启、关闭过程。

对于优雅重启：

不关闭现有连接（正在运行中的程序）
新的进程启动并替代旧进程
新的进程接管新的连接
连接要随时响应用户的请求，当用户仍在请求旧进程时要保持连接，新用户应请求新进程，不可以出现拒绝请求的情况

对于优雅关闭：

先标记为不接收和向下游发送新请求，新请求过来时直接报错，让客户端重试其它机器。
程序中是否还有关键在运行（请求过来触发的逻辑、自身循环逻辑、定时任务逻辑等），如果有，等待上有请求触发的逻辑执行完成，如果超时，强制取消，对于自身内部的一些逻辑，通过上下文发送取消动作，如果超时，强制执行关闭。

系统信号

通过信号通知服务重启、关闭

Linux 62个
1) SIGHUP	 2) SIGINT	 3) SIGQUIT	 4) SIGILL	 5) SIGTRAP
 6) SIGABRT	 7) SIGBUS	 8) SIGFPE	 9) SIGKILL	10) SIGUSR1
11) SIGSEGV	12) SIGUSR2	13) SIGPIPE	14) SIGALRM	15) SIGTERM
16) SIGSTKFLT	17) SIGCHLD	18) SIGCONT	19) SIGSTOP	20) SIGTSTP
21) SIGTTIN	22) SIGTTOU	23) SIGURG	24) SIGXCPU	25) SIGXFSZ
26) SIGVTALRM	27) SIGPROF	28) SIGWINCH	29) SIGIO	30) SIGPWR
31) SIGSYS	34) SIGRTMIN	35) SIGRTMIN+1	36) SIGRTMIN+2	37) SIGRTMIN+3
38) SIGRTMIN+4	39) SIGRTMIN+5	40) SIGRTMIN+6	41) SIGRTMIN+7	42) SIGRTMIN+8
43) SIGRTMIN+9	44) SIGRTMIN+10	45) SIGRTMIN+11	46) SIGRTMIN+12	47) SIGRTMIN+13
48) SIGRTMIN+14	49) SIGRTMIN+15	50) SIGRTMAX-14 51) SIGRTMAX-13 52) SIGRTMAX-12
53) SIGRTMAX-11	54) SIGRTMAX-10 55) SIGRTMAX-9	56) SIGRTMAX-8	57) SIGRTMAX-7
58) SIGRTMAX-6	59) SIGRTMAX-5	60) SIGRTMAX-4	61) SIGRTMAX-3	62) SIGRTMAX-2
63) SIGRTMAX-1	64) SIGRTMAX

macOS 31个
HUP INT QUIT ILL TRAP ABRT EMT FPE KILL BUS SEGV SYS PIPE ALRM TERM URG STOP TSTP CONT CHLD TTIN TTOU IO XCPU XFSZ VTALRM PROF WINCH INFO USR1 USR2

windows 7个,  在signal.h中定义
SIGINT      Ctrl+C中断
SIGILL       非法指令
SIGFPE      浮点异常
SIGSEGV   段错误, 非法指针访问
SIGTERM   kill发出的软件终止
SIGBREAK Ctrl+Break中断
SIGABRT   调用abort导致

操作对应的信号

例子

服务关闭

package main

import (
	"context"
	"fmt"
	"math/rand"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

const (
	TimeTemplate = "15:04:05.999999999"
)

type Service interface {
	GetName() string
	Serve(ctx context.Context)
	Shutdown() error
}

type BusinessService struct {
}

func (b *BusinessService) GetName() string {
	return "BusinessService"
}

func (b *BusinessService) Serve(ctx context.Context) {
	for {
		fmt.Printf("BusinessService serve run at %s\n", time.Now().Format(TimeTemplate))
		select {
		case <-ctx.Done():
			fmt.Printf("BusinessService serve done at %s\n", time.Now().Format(TimeTemplate))
			return
		default:
			if n := rand.Intn(10); n > 5 {
				panic(fmt.Errorf("make panic from BusinessService by reason: random panic on %d", n))
			}
		}
		time.Sleep(time.Second)
	}
	return
}

func (b *BusinessService) Shutdown() error {
	fmt.Printf("BusinessService shutdown begin... at %s\n", time.Now().Format(TimeTemplate))
	//todo destory
	defer func() {
		fmt.Printf("BusinessService shutdown end... at %s\n", time.Now().Format(TimeTemplate))
	}()
	return nil
}

type LogService struct {
	buffer []string
}

func (l *LogService) Serve(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			fmt.Printf("LogService serve done at %s\n", time.Now().Format(TimeTemplate))
			return
		default:
			// 0.5s append one log
			time.Sleep(500 * time.Millisecond)
			l.buffer = append(l.buffer, fmt.Sprintf("Time: %d", time.Now().Unix()))
		}
	}
}

func (b *LogService) GetName() string {
	return "LogService"
}

func (l *LogService) Shutdown() (err error) {
	fmt.Printf("LogService shutdown begin... at %s\n", time.Now().Format(TimeTemplate))
	defer fmt.Printf("LogService shutdown end... at %s\n", time.Now().Format(TimeTemplate))
	if len(l.buffer) == 0 {
		return
	}
	fmt.Printf("cache [%d] wait to send \n", len(l.buffer))
	for _, log := range l.buffer {
		fmt.Printf("send Log [%s]\n", log)
	}
	return
}

type ServiceGroup struct {
	ctx      context.Context
	cancel   func()
	services []Service //service list
}

func NewServiceGroup(ctx context.Context) *ServiceGroup {
	g := ServiceGroup{}
	g.ctx, g.cancel = context.WithCancel(ctx)
	return &g
}

func (s *ServiceGroup) Add(service Service) {
	s.services = append(s.services, service)
}

func (s *ServiceGroup) run(service Service) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = r.(error)
			fmt.Printf("receive panic msg: %s\n", err.Error())
		}
	}()
	//with cancel ctx to child context
	service.Serve(s.ctx)
	return
}

func (s *ServiceGroup) watchDog() {
	signalChan := make(chan os.Signal, 1)
	signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
	for {
		select {
		case signalData := <-signalChan:
			switch signalData {
			case syscall.SIGINT:
				fmt.Println("receive signal sigint")
			case syscall.SIGTERM:
				fmt.Println("receive signal sigerm")
			default:
				fmt.Println("receive singal unknown")
			}
			// do cancel notify all services cancel
			s.cancel()
			goto CLOSE
		case <-s.ctx.Done():
			goto CLOSE
		}
	}
CLOSE:
	for _, service := range s.services {
		if err := service.Shutdown(); err != nil {
			fmt.Printf("shutdown failed err: %s", err)
		}
	}
}

func (s *ServiceGroup) ServeAll() {
	var wg sync.WaitGroup
	for idx := range s.services {
		service := s.services[idx]
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := s.run(service); err != nil {
				fmt.Printf("receive service [%s] has error: 【%s】, do cancel\n", service.GetName(), err.Error())
				s.cancel()
			}
		}()
	}
	wg.Add(1)
	go func() {
		defer wg.Done()
		s.watchDog()
	}()
	wg.Wait()
}

func main() {
	rand.Seed(time.Now().Unix())
	ctx := context.Background()

	g := NewServiceGroup(ctx)
	g.Add(&LogService{})
	g.Add(&BusinessService{})
	g.ServeAll()
}

服务重启

facebook的重启实现，类似endless

// Command gracedemo implements a demo server showing how to gracefully
// terminate an HTTP server using grace.
package main

import (
	"fmt"
	"net/http"
	"os"
	"time"

	"github.com/facebookgo/grace/gracehttp"
)

var (
	now = time.Now()
)

func main() {
	fmt.Printf("server start pid:%d", os.Getpid())
	gracehttp.Serve(
		&http.Server{Addr: ":1111", Handler: newHandler("Zero  ")},
		&http.Server{Addr: ":1112", Handler: newHandler("First ")},
		&http.Server{Addr: ":1113", Handler: newHandler("Second")},
	)
}

func newHandler(name string) http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("/sleep/", func(w http.ResponseWriter, r *http.Request) {
		duration, err := time.ParseDuration(r.FormValue("duration"))
		if err != nil {
			http.Error(w, err.Error(), 400)
			return
		}
		time.Sleep(duration)
		fmt.Fprintf(
			w,
			"%s started at %s slept for %d nanoseconds from pid %d.\n",
			name,
			now,
			duration.Nanoseconds(),
			os.Getpid(),
		)
	})
	return mux
}

// curl 'http://localhost:1111/sleep/?duration=0s'
// curl 'http://localhost:1111/sleep/?duration=30s'
// kill -USR2 14642

业界的实现

nginx实现服务重启

获取进程id.   cat /usr/local/nginx/logs/nginx.pid
优雅重启. kill -HUP (进程号) 例：kill -HUP 'cat /usr/local/nginx/logs/nginx.pid'
优雅停止 kill -QUIT (进程号)
暴力停止 kill -TERM (进程号) kill -INT (进程号)

其他信号指令
kill -USR1 (进程号)   //重读日志
kill -USR2 (进程号)   //平滑升级
kill -WINCH (进程号)  //优雅关闭旧的进程，配合USR2

强制停止 pkill -9 nginx

使用nginx -s reload进行平滑重启。nginx启动时会通过参数-s 发现目前要进行信号处理而不是启动nginx服务，然后他会查看nginx的pid文件，pid文件中保存有master的进程号，然后向master进行发送相应的信号，reload对应的是HUP信号，所以nginx –s reload 跟kill -1 masterpid 一样。Master收到HUP信号后的处理流程如下：

1）master解析新的配置文件。

2）master fork出新的worker进程，此时新的worker会和旧的worker共存。

3）master向旧的worker发送QUIT命令。

4）旧的worker会关闭监听端口，不再接受新的网络请求，并等待所有正在处理的请求完成后，退出。

5）此时只有新的worker存在，nginx完成了重启。

类型	效果
grace	旧API不会断掉，会执行原来的逻辑，pid会变化
endless	旧API不会断掉，会执行原来的逻辑，pid会变化
overseer	旧API不会断掉，会执行原来的逻辑，主进程pid不会变化

三种类型，其中nginx类似overseer，主进程pid不变化，work进程变化。

grace和endless是比较类似。

监听信号

收到信号时fork子进程（使用相同的启动命令），将服务监听的socket文件描述符传递给子进程

子进程监听父进程的socket，这个时候父进程和子进程都可以接收请求

子进程启动成功之后，父进程停止接收新的连接，等待旧连接处理完成（或超时）

父进程退出，升级完成

overseer是不同的，主要是overseer加了一个主进程管理平滑重启，子进程处理链接，能够保持主进程pid不变，与nginx类似。