1.簡介

本文將介紹首先為什么需要主動關(guān)閉goroutine，并介紹如何在Go語言中關(guān)閉goroutine的常見套路,包括傳遞終止信號和協(xié)程內(nèi)部捕捉終止信號。之后，文章列舉了需要主動關(guān)閉協(xié)程運(yùn)行的常見場景，如啟動一個協(xié)程執(zhí)行一個不斷重復(fù)的任務(wù)。希望通過本文的介紹，讀者能夠掌握如何在適當(dāng)?shù)臅r候關(guān)閉goroutine，以及了解關(guān)閉goroutine的常見套路。

2.為什么需要關(guān)閉goroutine

2.1 協(xié)程的生命周期

了解協(xié)程的生命周期是優(yōu)雅地關(guān)閉協(xié)程的前提，因為在關(guān)閉協(xié)程之前需要知道協(xié)程的當(dāng)前狀態(tài)，以便采取相應(yīng)的措施。所以這里我們需要先了解下goroutine的生命周期。

在 Go語言中，協(xié)程（goroutine）是一種輕量級的線程，可以在一個程序中同時運(yùn)行多個協(xié)程，提高程序的并發(fā)性能。協(xié)程的生命周期包括創(chuàng)建、運(yùn)行和結(jié)束三個階段。

首先需要創(chuàng)建一個協(xié)程，協(xié)程的創(chuàng)建可以通過關(guān)鍵字 go 來實現(xiàn)，例如：

go func() {
    // 協(xié)程執(zhí)行的代碼
}()

上面的代碼會啟動一個新的協(xié)程，同時在新的協(xié)程中執(zhí)行匿名函數(shù)，此時協(xié)程便已被創(chuàng)建了。

一旦協(xié)程被創(chuàng)建，它就會在新的線程中運(yùn)行。協(xié)程的運(yùn)行狀態(tài)可以由 Go 運(yùn)行時（goroutine scheduler）來管理，它會自動將協(xié)程調(diào)度到適當(dāng)?shù)?code>P中運(yùn)行，并確保協(xié)程的公平調(diào)度和平衡負(fù)載。

在運(yùn)行階段，協(xié)程會不斷地執(zhí)行任務(wù)，直到任務(wù)完成或者遇到終止條件。在終止階段，協(xié)程將會被回收，從而完成其整個生命周期。

綜上所述，協(xié)程由go關(guān)鍵字啟動，在協(xié)程中執(zhí)行其業(yè)務(wù)邏輯，直到最后遇到終止條件，此時代表著協(xié)程的任務(wù)已經(jīng)結(jié)束了，將進(jìn)入終止階段。最終協(xié)程將會被回收。

2.2 協(xié)程的終止條件

正常來說，都是協(xié)程任務(wù)執(zhí)行完成之后，此時協(xié)程自動退出，例如:

func main() {
   var wg sync.WaitGroup
   wg.Add(1)
   go func() {
      defer wg.Done()
      // 協(xié)程執(zhí)行的代碼
      fmt.Println("協(xié)程執(zhí)行完畢")
   }()
   wg.Wait()
   // 等待協(xié)程執(zhí)行完畢
   fmt.Println("主程序結(jié)束")

上面的代碼中，我們使用 WaitGroup 等待協(xié)程執(zhí)行完畢。在協(xié)程執(zhí)行完畢后，程序會輸出協(xié)程執(zhí)行完畢和主程序結(jié)束兩條信息。

還有一種情況是協(xié)程發(fā)生panic，它將會自動退出。例如：

func main() {
    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        defer wg.Done()
        // 協(xié)程執(zhí)行的代碼
        panic("協(xié)程發(fā)生錯誤")
    }()
    // 等待協(xié)程執(zhí)行完畢
    wg.Wait()
    fmt.Println("主程序結(jié)束")
}

在這種情況下，協(xié)程也會自動退出，不會再占用系統(tǒng)資源。

綜合看來，協(xié)程的終止條件，其實就是協(xié)程中的任務(wù)執(zhí)行完成了，或者是執(zhí)行過程中發(fā)生了panic，協(xié)程將滿足終止條件，退出執(zhí)行。

2.3 為什么需要主動關(guān)閉goroutine

從上面協(xié)程的終止條件來看，正常情況下，協(xié)程只要將任務(wù)正常處理完成，協(xié)程自動退出，此時并不需要主動關(guān)閉goroutine。

這里先舉一個生產(chǎn)者消費(fèi)者的例子，在這個例子中，我們創(chuàng)建了一個生產(chǎn)者和一個消費(fèi)者，它們之間通過一個channel進(jìn)行通信。生產(chǎn)者生產(chǎn)數(shù)據(jù)并發(fā)送到一個channel中，消費(fèi)者從這個channel中讀取數(shù)據(jù)并進(jìn)行處理。代碼示例如下:

func main() {
    // 生產(chǎn)者代碼
    go func(out chan<- int) {
        for i := 0; ; i++ {
            select {
            case out <- i:
                fmt.Printf("producer: produced %d\n", i)
            time.Sleep(time.Second)
        }
    }
    // 消費(fèi)者邏輯
    go func(in <-chan int) {
        for {
            select {
            case i := <-in:
                fmt.Printf("consumer: consumed %d\n", i)
            }
        }
    }
    // 讓生產(chǎn)者協(xié)程和消費(fèi)者協(xié)程一直執(zhí)行下去
    time.Sleep(100000000)
}

在這個例子中，我們使用了兩個goroutine：生產(chǎn)者和消費(fèi)者。生產(chǎn)者向channel中生產(chǎn)數(shù)據(jù)，消費(fèi)者從channel中消費(fèi)數(shù)據(jù)。

但是，假如生產(chǎn)者出現(xiàn)了問題，此時生產(chǎn)者的協(xié)程將會被退出，不再執(zhí)行。而消費(fèi)者仍然在等待數(shù)據(jù)的輸入。此時消費(fèi)者協(xié)程已經(jīng)沒有存在的必要了，其實是需要退出執(zhí)行。

因此，對于一些雖然沒有達(dá)到終止條件的協(xié)程，但是其又沒有再繼續(xù)執(zhí)行下去的必要，此時主動關(guān)閉其執(zhí)行，從而保證程序的健壯性和性能。

3.如何優(yōu)雅得關(guān)閉goroutine

優(yōu)雅得關(guān)閉goroutine的執(zhí)行，我們可以遵循以下三個步驟。首先是傳遞關(guān)閉協(xié)程的信號，其次是協(xié)程內(nèi)部需要能夠到關(guān)閉信號，最后是協(xié)程退出時，能夠正確釋放其所占據(jù)的資源。通過以上步驟，可以保在需要時優(yōu)雅地停止goroutine的執(zhí)行。下面對這三個步驟詳細(xì)進(jìn)行講解。

3.1 傳遞關(guān)閉終止信號

首先是通過給goroutine傳遞關(guān)閉協(xié)程的信號，從而讓協(xié)程進(jìn)行退出操作。這里可以使用context.Context來傳遞信號，具體實現(xiàn)可以通過調(diào)用WithCancel,WithDeadline,WithTimeout等方法來創(chuàng)建一個帶有取消功能的Context，并在需要關(guān)閉協(xié)程時調(diào)用Cancel方法來向Context發(fā)送取消信號。示例代碼如下：

ctx, cancel := context.WithCancel(context.Background())
go func(ctx context.Context) {
    for {
        select {
        // 調(diào)用cancel函數(shù)后,這里將能夠收到通知
        case <-ctx.Done():
            return
        default:
            // do something
        }
    }
}(ctx)
// 在需要關(guān)閉協(xié)程時調(diào)用cancel方法發(fā)送取消信號
cancel()

這里，當(dāng)我們想要終止協(xié)程的執(zhí)行時，只需要調(diào)用可取消context對象的Cancel方法，協(xié)程內(nèi)部將能夠通過context對象接收到終止協(xié)程執(zhí)行的通知。

3.2 協(xié)程內(nèi)部捕捉終止信號

  協(xié)程內(nèi)部也需要在取消信號傳遞過來時，能夠正確被捕捉到，才能夠正常終止流程。這里我們可以使用select語句來監(jiān)聽取消信號。select語句可以有多個case子句，可以同時監(jiān)聽多個channel，當(dāng)select語句執(zhí)行時，它會一直阻塞，直到有一個case子句可以執(zhí)行。select語句也可以包含default子句，這個子句在所有的case子句都不能執(zhí)行時會被執(zhí)行，通常用于防止select語句的阻塞。如下:

select {
case <-channel:
    // channel有數(shù)據(jù)到來時執(zhí)行的代碼
default:
    // 所有channel都沒有數(shù)據(jù)時執(zhí)行的代碼
}

而context對象的Done方法剛好也是返回一個channel,取消信號便是通過該channel來進(jìn)行傳遞的。所以我們可以在協(xié)程內(nèi)部，通過select語句，在其中一個case分支來監(jiān)聽取消信號；同時使用一個default分支在協(xié)程中執(zhí)行具體的業(yè)務(wù)邏輯。在終止信號沒有到來時，就執(zhí)行業(yè)務(wù)邏輯；在收到協(xié)程終止信號后，也能夠及時終止協(xié)程的執(zhí)行。如下:

go func(ctx context.Context) {
    for {
        select {
        // 調(diào)用cancel函數(shù)后,這里將能夠收到通知
        case <-ctx.Done():
            return
        default:
            // 執(zhí)行業(yè)務(wù)邏輯
        }
    }
}(ctx)

3.3 回收協(xié)程資源

最后，當(dāng)協(xié)程被終止執(zhí)行時，需要釋放占用的資源，包括文件句柄、內(nèi)存等，以便其他程序可以繼續(xù)使用這些資源。在Go語言中，可以使用defer語句來確保協(xié)程在退出時能夠正確地釋放資源。比如協(xié)程中打開了一個文件，此時可以通過defer語句來關(guān)閉，避免資源的泄漏。代碼示例如下:

func doWork() {
    file, err := os.Open("test.txt")
    if err != nil {
        log.Fatal(err)
    }
    defer file.Close()

    // Do some work
}

在這個例子中，我們在文件打開之后使用defer語句注冊了一個函數(shù)，當(dāng)協(xié)程結(jié)束時會自動調(diào)用該函數(shù)來關(guān)閉文件。這樣協(xié)程無論在何時退出，我們都可以確保文件被正確關(guān)閉，避免資源泄漏和其他問題。

3.4 關(guān)閉goroutine示例

下面展示一個簡單的例子，結(jié)合Context對象，select語句以及defer語句這三部分內(nèi)容，優(yōu)雅得終止一個協(xié)程的運(yùn)行，具體代碼示例如下:

package main

import (
    "context"
    "fmt"
    "time"
)

func worker(ctx context.Context) {
    // 最后,在協(xié)程退出前,釋放資源.
    defer fmt.Println("worker stopped")

    for {
        // 通過select語句監(jiān)聽取消信號,取消信號沒到達(dá),則執(zhí)行業(yè)務(wù)邏輯,等下次循環(huán)檢查
        select {
        default:
            fmt.Println("working")
        case <-ctx.Done():
            return
        }
        time.Sleep(time.Second)
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    // 啟動一個協(xié)程執(zhí)行任務(wù)
    go worker(ctx)
    // 執(zhí)行5s后,調(diào)用cancel函數(shù)終止協(xié)程
    time.Sleep(5 * time.Second)
    cancel()

    time.Sleep(2 * time.Second)
}

在main函數(shù)中，我們使用context.WithCancel函數(shù)創(chuàng)建了一個新的context，并將其傳遞給worker函數(shù)，同時啟動協(xié)程運(yùn)行worker函數(shù)。

當(dāng)worker函數(shù)執(zhí)行5s后，主協(xié)程調(diào)用cancel函數(shù)來終止worker協(xié)程。之后，worker協(xié)程中監(jiān)聽取消信號的select語句，將能夠捕捉到這個信號，執(zhí)行終止協(xié)程操作。

最后，在退出協(xié)程時，通過defer語句實現(xiàn)資源的釋放。綜上，我們實現(xiàn)了協(xié)程的優(yōu)雅關(guān)閉，同時也正確回收了資源。

4. 需要主動關(guān)閉協(xié)程運(yùn)行的常見場景

4.1 協(xié)程在執(zhí)行一個不斷重復(fù)的任務(wù)

協(xié)程在執(zhí)行一個不斷重復(fù)的任務(wù)時，此時協(xié)程是不會主動終止運(yùn)行的。但是在某個時刻之后，不需要再繼續(xù)執(zhí)行該任務(wù)了，需要主動關(guān)閉goroutine的執(zhí)行，釋放協(xié)程的資源。

這里以etcd為例來進(jìn)行說明。etcd主要用于在分布式系統(tǒng)中存儲配置信息、元數(shù)據(jù)和一些小規(guī)模的共享數(shù)據(jù)。也就是說，我們可以在etcd當(dāng)中存儲一些鍵值對。那么，如果我們想要設(shè)置鍵值對的有效期，那該如何實現(xiàn)呢?

etcd中存在一個租約的概念，租約可以看作是一個時間段，該時間段內(nèi)某個鍵值對的存在是有意義的，而在租約到期后，該鍵值對的存在便沒有意義，可以被刪除，同時一個租約可以作用于多個鍵值對。下面先展示如何將一個租約和一個key進(jìn)行關(guān)聯(lián)的示例：

// client 為 etcd客戶端的連接,基于此建立一個Lease實例
// Lease示例提供一些api,能過創(chuàng)建租約，取消租約,續(xù)約租約
lease := clientv3.NewLease(client)

// 創(chuàng)建一個租約,同時租約時間為10秒
grantResp, err := lease.Grant(context.Background(), 10)
if err != nil {
    log.Fatal(err)
}
// 租約ID,每一個租約都有一個唯一的ID
leaseID := grantResp.ID

// 將租約與key進(jìn)行關(guān)聯(lián),此時該key的有效期,也就是該租約的有效期
_, err = kv.Put(context.Background(), "key1", "value1", clientv3.WithLease(leaseID))
if err != nil {
    log.Fatal(err)
}

以上代碼演示了如何在etcd中創(chuàng)建一個租約并將其與一個鍵值對進(jìn)行關(guān)聯(lián)。首先，通過etcd客戶端的連接創(chuàng)建了一個Lease實例，該實例提供了一些api，可以創(chuàng)建租約、取消租約和續(xù)約租約。然后使用Grant函數(shù)創(chuàng)建了一個租約并指定了租約的有效期為10秒。接下來，獲取租約ID，每個租約都有一個唯一的ID。最后，使用Put函數(shù)將租約與key進(jìn)行關(guān)聯(lián)，從而將該key的有效期設(shè)定為該租約的有效期。

所以，我們?nèi)绻胍僮?code>etcd中鍵值對的有效期，只需要操作租約的有效期即可。

而剛好，etcd其實定義了一個Lease接口，該接口定義了對租約的一些操作，能過創(chuàng)建租約，取消租約，同時也支持續(xù)約租約，獲取過期時間等內(nèi)容，具體如下:

type Lease interface {
   // 1. 創(chuàng)建一個新的租約
   Grant(ctx context.Context, ttl int64) (*LeaseGrantResponse, error)
   // 2. 取消租約
   Revoke(ctx context.Context, id LeaseID) (*LeaseRevokeResponse, error)
   // 3. 獲取租約的剩余有效期
   TimeToLive(ctx context.Context, id LeaseID, opts ...LeaseOption) (*LeaseTimeToLiveResponse, error)
   // 4. 獲取所有的租約
   Leases(ctx context.Context) (*LeaseLeasesResponse, error)
   // 5. 不斷對租約進(jìn)行續(xù)約，這里假設(shè)10s后過期，此時大概的含義為每隔10s續(xù)約一次租約,調(diào)用該方法后,租約將永遠(yuǎn)不會過期  
   KeepAlive(ctx context.Context, id LeaseID) (<-chan *LeaseKeepAliveResponse, error)
   // 6. 續(xù)約一次租約
   KeepAliveOnce(ctx context.Context, id LeaseID) (*LeaseKeepAliveResponse, error)
   // 7. 關(guān)閉Lease實例 
   Close() error
}

到此為止，我們引出了Lease接口，而其中KeepAlive方法便是我們今日的主角，從該方法定義可以看出，當(dāng)調(diào)用KeepAlive方法對某個租約進(jìn)行續(xù)約后，其每隔一段時間都會執(zhí)行對目標(biāo)租約的續(xù)約操作。這個時候一般都是啟動一個協(xié)程，由協(xié)程來完成對租約的續(xù)約操作。

此時協(xié)程其實就是在執(zhí)行一個不斷重復(fù)的任務(wù)，那如果Lease接口的實例調(diào)用了Close方法，想要回收掉Lease實例，不會再通過該實例對租約進(jìn)行操作，回收掉Lease所有占據(jù)的資源，那么KeepAlive方法創(chuàng)建的協(xié)程，此時也應(yīng)該被主動關(guān)閉，不應(yīng)該再繼續(xù)執(zhí)行下去。

事實上，當(dāng)前etcd中Lease接口中KeepAlive方法的默認(rèn)實現(xiàn)也是如此。并且對主動關(guān)閉協(xié)程運(yùn)行的實現(xiàn)，也是通過context傳遞對象，select獲取取消信號，最后通過defer 來回收資源這三者組合起來實現(xiàn)的。

下面來看看執(zhí)行續(xù)約操作的函數(shù)，會啟動一個協(xié)程在后臺不斷執(zhí)行，具體實現(xiàn)如下:

func (l *lessor) sendKeepAliveLoop(stream pb.Lease_LeaseKeepAliveClient) {
   for {
      var tosend []LeaseID
      
      now := time.Now()
      l.mu.Lock()
      // keepAlives 是保存了所有待續(xù)約的 租約ID
      for id, ka := range l.keepAlives {
         // 然后nextKeepAlive為下次續(xù)約的時間,如果超過該時間,則執(zhí)行續(xù)約操作
         if ka.nextKeepAlive.Before(now) {
            tosend = append(tosend, id)
         }
      }
      l.mu.Unlock()
      // 發(fā)送續(xù)約請求
      for _, id := range tosend {
         r := &pb.LeaseKeepAliveRequest{ID: int64(id)}
         // 向etcd集群發(fā)送續(xù)約請求
         if err := stream.Send(r); err != nil {
            return
         }
      }

      select {
      // 每隔500ms執(zhí)行一次
      case <-time.After(500 * time.Millisecond):
      // 如果接收到終止信號,則直接終止
      case <-l.stopCtx.Done():
         return
      }
   }
}

可以看到，其會不斷循環(huán)，首先會檢查當(dāng)前時間是否超過了所有租約的下次續(xù)約時間，如果超過了，則會將這些租約的 ID 放入 tosend 數(shù)組中，并在循環(huán)的下一步中向 etcd集群發(fā)送續(xù)約請求。接著會等待 500 毫秒，然后再次執(zhí)行上述操作。正常情況下，其不會退出循環(huán)，會一直向etcd集群發(fā)送續(xù)約請求。除非收到了終止信號，其才會退出，從而正常結(jié)束協(xié)程。

而stopCtx則是lessor實例的變量，用于傳遞取消信號。在創(chuàng)建 lessor 實例時，stopCtx 是由 context.WithCancel() 函數(shù)創(chuàng)建的。這個函數(shù)會返回兩個對象：一個帶有取消方法的 context.Context 對象（即 stopCtx），以及一個函數(shù)對象 stopCancel，調(diào)用這個函數(shù)會取消上下文對象。具體如下:

// 創(chuàng)建Lease實例
func NewLeaseFromLeaseClient(remote pb.LeaseClient, c *Client, keepAliveTimeout time.Duration) Lease {
   // ...省略一些無關(guān)內(nèi)容
   reqLeaderCtx := WithRequireLeader(context.Background())
   // 通過withCancel函數(shù)創(chuàng)建cancelCtx對象
   l.stopCtx, l.stopCancel = context.WithCancel(reqLeaderCtx)
   return l
}

在 lessor.Close() 函數(shù)中，我們調(diào)用 stopCancel() 函數(shù)來發(fā)送取消信號。

func (l *lessor) Close() error {
   l.stopCancel()
   // close for synchronous teardown if stream goroutines never launched
   // 省略無關(guān)內(nèi)容
   return nil
}

因為 sendKeepAliveLoop() 協(xié)程會在 stopCtx 上等待信號，所以一旦調(diào)用了 stopCancel()，協(xié)程會收到信號并退出。這個機(jī)制非常靈活，因為stopCtx是實例的成員變量，所以lessor實例創(chuàng)建的所有協(xié)程，都可以通過監(jiān)聽stopCtx來決定是否要退出執(zhí)行。

5.總結(jié)

這篇文章主要介紹了為什么需要主動關(guān)閉goroutine，以及在Go語言中關(guān)閉goroutine的常見套路。

文章首先介紹了為什么需要主動關(guān)閉goroutine。接下來，文章詳細(xì)介紹了Go語言中關(guān)閉goroutine的常見套路，包括傳遞終止信號和協(xié)程內(nèi)部捕捉終止信號。在傳遞終止信號的方案中，文章介紹了如何使用context對象傳遞信號，并使用select語句等待信號。在協(xié)程內(nèi)部捕捉終止信號的方案中，文章介紹了如何使用defer語句來回收資源。

最后，文章列舉了需要主動關(guān)閉協(xié)程運(yùn)行的常見場景，如協(xié)程在執(zhí)行一個不斷重復(fù)的任務(wù)，在不再需要繼續(xù)執(zhí)行下去的話，就需要主動關(guān)閉協(xié)程的執(zhí)行。希望通過本文的介紹，讀者能夠掌握如何在適當(dāng)?shù)臅r候關(guān)閉goroutine，從而避免資源浪費(fèi)的問題。

以上就是Go如何優(yōu)雅的關(guān)閉goroutine協(xié)程的詳細(xì)內(nèi)容，更多關(guān)于Go 關(guān)閉goroutine的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論