造輪子必備：什么是優(yōu)雅關(guān)閉？

菌心說 2021-11-21

展開全文

引子?

關(guān)閉?還特么優(yōu)雅?

實際上優(yōu)雅關(guān)閉還有另外一個名詞, 叫“平滑退出'。如果你打算自己造輪子, 優(yōu)雅關(guān)閉將是你要掌握的第一個知識點。

在生活中，如果有一個名詞確實過于難以理解，我們不妨來看這個名詞的反面是什么。

舉個簡單的例子

優(yōu)雅關(guān)閉: 就是使用操作系統(tǒng)關(guān)機功能關(guān)閉你的計算機
不太優(yōu)雅的關(guān)閉: 直接斷電重啟

可能有同學會有疑問了: 我平時電腦卡住的時候都是直接斷電重啟的啊，也沒見有啥大問題啊?

計算機與計算機的體質(zhì)不能一概而論, 如果你的計算機安裝的是Linux系統(tǒng)，恰好這又是一臺服務(wù)器的話，強制重啟的話，你大概率會丟掉部分數(shù)據(jù)，如果是生產(chǎn)環(huán)境的話，那就準備提桶跑路吧。

為什么呢?

如果你曾經(jīng)想過要做MySQL或者Redis的調(diào)優(yōu), 或多或少接觸過以下參數(shù):

MySQL的sync_binlog參數(shù), 用來控制持久化binlog數(shù)據(jù)的存儲設(shè)備的行為
Redis的appendfysnc參數(shù), 用來控制Redis持久化AOF日志到數(shù)據(jù)存儲設(shè)備的行為。

出現(xiàn)以上參數(shù)的原因是因為將數(shù)據(jù)持久化到存儲設(shè)備是一個耗時相對較高的行為, Linux采取的優(yōu)化措施是，當你往一個文件中數(shù)據(jù)時會暫存到系統(tǒng)的緩存中, 等待時機再批量持久化到存儲設(shè)備中。除非進程指定使用DirectIO的方式或者調(diào)用fsync，操作系統(tǒng)才會主動將數(shù)據(jù)寫入存儲設(shè)備。

因此MySQL和Redis紛紛開放了調(diào)優(yōu)參數(shù)用來控制日志持久化行為，并將鍋甩回給了程序員。

sync_binlog 的值一般為1, 即每次提交事務(wù)就立即將binlog數(shù)據(jù)持久化到存儲設(shè)備。

優(yōu)雅關(guān)閉

現(xiàn)在，從不太優(yōu)雅關(guān)閉的例子了解到優(yōu)雅關(guān)閉要做什么了:

讓程序完成未完成的工作(如：提交事務(wù), 持久化日志等等)

但是，我們還需要加一個限制條件:

當程序決定優(yōu)雅關(guān)閉的時候，就不能再接送任何請求。

不停止處理新請求的話就永遠沒完沒了

線程池的優(yōu)雅關(guān)閉

線程池(ThreadPoolExecutor)在JDK的并發(fā)包中占據(jù)了重要的位置, 我們可以來看看如此重要的一個基礎(chǔ)組件是如何處理優(yōu)雅關(guān)閉的。

該類將是否需要優(yōu)雅關(guān)閉的權(quán)限開放給程序員, 并提供了兩個方法，分別是:

ThreadPoolExecutor.shutdown 該方法會將線程池的狀態(tài)設(shè)置為SHUTDOWN, 并且不再接受新任務(wù)的提交，但會讓線程池中的線程跑完所有已提交的任務(wù)
ThreadPoolExecutor.shutdownNow 該方法會將線程池的狀態(tài)設(shè)置為STOP, 并且不再接受新任務(wù)的提交, 以及立即向線程池中的所有線程發(fā)出中斷信號，對于使已提交到線程池中但還未運行的任務(wù)直接忽視掉。

優(yōu)雅關(guān)閉進程

如何優(yōu)雅關(guān)閉進程呢? 首先我們需要搞清楚進程什么情況下會關(guān)閉:

主動關(guān)閉(對于一個對外提供服務(wù)的進程來說通常不會主動關(guān)閉)
程序崩潰, 如某個業(yè)務(wù)拋出異常處理不當，導(dǎo)致異常拋到最外層并且沒有進行處理導(dǎo)致程序崩潰
進程收到來自操作系統(tǒng)的關(guān)閉信號(如按下Ctrl+C)

在企業(yè)級應(yīng)用中，一個進程通常不止有業(yè)務(wù)邏輯，還有圍繞著業(yè)務(wù)而開發(fā)的日志服務(wù)/MQ服務(wù)/運維服務(wù)等等, 那么當某個業(yè)務(wù)出現(xiàn)可能導(dǎo)致進程崩潰的問題時，我們就需要將進程即將關(guān)閉的消息廣播給其他服務(wù), 并調(diào)用這些服務(wù)提供的優(yōu)雅關(guān)閉方法，以上措施全部完成后再退出進程, 如日志服務(wù)的優(yōu)雅關(guān)閉是確保日志落盤, MQ服務(wù)的優(yōu)雅關(guān)閉是確保消息被投遞出去或者被消費完等等。

如果你是直接殺進程(kill -9)的話, 也就沒有必要討論優(yōu)雅關(guān)閉了

我們以Golang為例來描述如何優(yōu)雅關(guān)閉進程, 首先我們需要對進程中的服務(wù)做一個抽象，以便實現(xiàn)生命周期管理, 每個服務(wù)提供均需要提供Serve和Shutdown方法。

type Service interface { Serve(ctx context.Context) error Shutdown() error}復(fù)制代碼

接下來我們定義一個ServiceGroup用來管理Service生命周期, 當任意Service運行出錯或接收系統(tǒng)信號SIGINT(Ctrl+C觸發(fā))和SIGTREM(kill 不加參數(shù)), ServiceGroup將負責關(guān)閉關(guān)閉由此管理的Service并調(diào)用Shutdown方法。

type ServiceGroup struct {	ctx      context.Context	cancel   func()	services []Service}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)		}	}()	err = 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 <- signalChan:			// 接收到系統(tǒng)信號, 通知停止服務(wù)			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.Println('服務(wù)異常, 進入退出流程!')				s.cancel()			}		}()	}	wg.Add(1)	go func() {		defer wg.Done()		s.watchDog()	}()	wg.Wait()}復(fù)制代碼

接下來，我們定義一個會隨機panic的業(yè)務(wù)服務(wù)以及日志服務(wù)。

type BusinessService struct {}func (b *BusinessService) Serve(ctx context.Context) (err error) {	times := 0	for {		fmt.Printf('業(yè)務(wù)運行中 %d\n', times)		select {		case <- ctx.Done():			fmt.Printf('BusinessService receive cancel signal\n')			return		default:			if n := rand.Intn(256); n > 200 {				panic(fmt.Errorf('random panic on %d', n))			}		}		time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))		times++	}	return}func (b *BusinessService) Shutdown() error {	fmt.Println('業(yè)務(wù)服務(wù), 關(guān)閉!')	return nil}type LogService struct {	buffer []string}func (l *LogService) Serve(ctx context.Context) (err error) {	for {		select {		case <- ctx.Done():			return		default:			// 投遞日志到消息隊列			time.Sleep(time.Millisecond * time.Duration(rand.Intn(500)))			l.buffer = append(l.buffer, fmt.Sprintf('Time: %d', time.Now().Unix()))		}	}}func (l *LogService) Shutdown() (err error) {	fmt.Printf('日志服務(wù), 關(guān)閉! 有[%d]條日志待發(fā)送\n', len(l.buffer))	if len(l.buffer) == 0 {		return	}	for _, log := range l.buffer {		// 發(fā)送日志或者持久化到硬盤		fmt.Printf('Send Log [%s]\n', log)	}	fmt.Println('緩沖區(qū)日志清理完畢')	return}復(fù)制代碼

運行

func main() {	rand.Seed(time.Now().Unix())	ctx := context.Background()	g := NewServiceGroup(ctx)	g.Add(&LogService{})	g.Add(&BusinessService{})	g.ServeAll()}復(fù)制代碼

運行輸出如下所示: