动态

详情 返回 返回

go語言控制goroutine協程退出的2種方法總結 - 动态 详情

我們知道,在go語言中,goroutine的執行會隨着main線程的退出而終結, 即如果main線程退出,則所有的goroutine都會被強制退出,不管你是否已經執行完畢。

如果我們希望main進程等待所有的goroutine執行完畢後再退出,則可以有3種方式來實現,具體如下:

1. 使用go標準庫sync中提供的 sync.WaitGroup裏面提供的Add, Done, Wait方法;

package main
 
import (
    "fmt"
    "sync"
    "time"
)
// 專業企業信息化軟件定製開發 免費諮詢 https://dev.tekin.cn/contactus.html
var wg sync.WaitGroup // 定義全局變量wg類型是sync.WaitGroup結構體, 因為我們要使用的方法是綁定在這個結構體上的
 
func test(n int) {
    defer wg.Done() //協程每次完成後執行這個將計數增量 -1; 注意這個代碼被調用的次數要和wg.Add(delta)這裏設置的增量一致
    for i := 1; i <= n; i++ {
        fmt.Printf("test %v \n", i)
        time.Sleep(100 * time.Millisecond)
    }
 
}
 
func main() {
 
    wg.Add(2) // 增加變量質量, 這裏的數字是你後面要啓動幾個協程就寫幾, 如要起2個協程就寫 2, 這裏的數字有1個餘量 即0, 所以如果是2 則wg.Done()最多可執行3次, 超過3次就會報panic異常, 如果 wg.Done()只執行1次則會報死鎖異常!!!
    go test(10)
    go test(5)
 
    test(6) // 這個正常 因為wg源碼裏面的增量比較是 < 0 所以
    //test(7) //這個會異常了 因為上面的的delta增量為2
 
    for i := 0; i < 10; i++ {
        fmt.Printf("main %v\n", i)
    }
    // 這裏會阻塞主進程等待所有的協程執行完畢後才會退出
    wg.Wait()
}

2. 利用管道chan讀取時會一直阻塞當前線程的特性實現等待

package main
 
import "fmt"
// 專業企業信息化軟件定製開發 免費諮詢 https://dev.tekin.cn/contactus.html
// 只讀/只寫 chan使用示例
// 發送消息 ch入參為僅寫
func Sender(ch chan<- int, exitCh chan struct{}) {
    for i := 0; i < 10; i++ {
        ch <- i
    }
    close(ch)
    var a struct{}
    exitCh <- a
}
 
// Receiver接收端 ch入參僅讀
func Receiver(ch <-chan int, exitCh chan struct{}) {
    //循環
    for {
        v, ok := <-ch
        if !ok {
            break //退出循環
        }
        fmt.Println("v=", v)
    }
    var a struct{}
    exitCh <- a
}
 
func main() {
    // 聲明sender chan
    var ch = make(chan int, 10)
    var exitCh = make(chan struct{}, 2)
    Sender(ch, exitCh)
    Receiver(ch, exitCh)
 
    var total = 0
    for _ = range exitCh {
        total++
        if total == 2 {
            break
        }
    }
    fmt.Println("結束...")
}

總結

上面2種方式, 第一種實現起來比較簡單,可少寫一些代碼, 但是性能相比第二種方式要低一些,因為第一種方式裏面使用了race,原子狀態維護和不少unsafe的方法(見後面的WaitGroup源碼參考)。 第二種方式代碼稍微複雜,但是效率較高,控制也比較靈活。

sync.WaitGroup源碼參考

// Copyright 2011 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
 
package sync
 
import (
    "internal/race"
    "sync/atomic"
    "unsafe"
)
 
// A WaitGroup waits for a collection of goroutines to finish.
// The main goroutine calls Add to set the number of
// goroutines to wait for. Then each of the goroutines
// runs and calls Done when finished. At the same time,
// Wait can be used to block until all goroutines have finished.
//
// A WaitGroup must not be copied after first use.
//
// In the terminology of the Go memory model, a call to Done
// “synchronizes before” the return of any Wait call that it unblocks.
type WaitGroup struct {
    noCopy noCopy
 
    state atomic.Uint64 // high 32 bits are counter, low 32 bits are waiter count.
    sema  uint32
}
 
// Add adds delta, which may be negative, to the WaitGroup counter.
// If the counter becomes zero, all goroutines blocked on Wait are released.
// If the counter goes negative, Add panics.
//
// Note that calls with a positive delta that occur when the counter is zero
// must happen before a Wait. Calls with a negative delta, or calls with a
// positive delta that start when the counter is greater than zero, may happen
// at any time.
// Typically this means the calls to Add should execute before the statement
// creating the goroutine or other event to be waited for.
// If a WaitGroup is reused to wait for several independent sets of events,
// new Add calls must happen after all previous Wait calls have returned.
// See the WaitGroup example.
func (wg *WaitGroup) Add(delta int) {
    if race.Enabled {
        if delta < 0 {
            // Synchronize decrements with Wait.
            race.ReleaseMerge(unsafe.Pointer(wg))
        }
        race.Disable()
        defer race.Enable()
    }
    state := wg.state.Add(uint64(delta) << 32)
    v := int32(state >> 32)
    w := uint32(state)
    if race.Enabled && delta > 0 && v == int32(delta) {
        // The first increment must be synchronized with Wait.
        // Need to model this as a read, because there can be
        // several concurrent wg.counter transitions from 0.
        race.Read(unsafe.Pointer(&wg.sema))
    }
    if v < 0 {
        panic("sync: negative WaitGroup counter")
    }
    if w != 0 && delta > 0 && v == int32(delta) {
        panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }
    if v > 0 || w == 0 {
        return
    }
    // This goroutine has set counter to 0 when waiters > 0.
    // Now there can't be concurrent mutations of state:
    // - Adds must not happen concurrently with Wait,
    // - Wait does not increment waiters if it sees counter == 0.
    // Still do a cheap sanity check to detect WaitGroup misuse.
    if wg.state.Load() != state {
        panic("sync: WaitGroup misuse: Add called concurrently with Wait")
    }
    // Reset waiters count to 0.
    wg.state.Store(0)
    for ; w != 0; w-- {
        runtime_Semrelease(&wg.sema, false, 0)
    }
}
 
// Done decrements the WaitGroup counter by one.
func (wg *WaitGroup) Done() {
    wg.Add(-1)
}
 
// Wait blocks until the WaitGroup counter is zero.
func (wg *WaitGroup) Wait() {
    if race.Enabled {
        race.Disable()
    }
    for {
        state := wg.state.Load()
        v := int32(state >> 32)
        w := uint32(state)
        if v == 0 {
            // Counter is 0, no need to wait.
            if race.Enabled {
                race.Enable()
                race.Acquire(unsafe.Pointer(wg))
            }
            return
        }
        // Increment waiters count.
        if wg.state.CompareAndSwap(state, state+1) {
            if race.Enabled && w == 0 {
                // Wait must be synchronized with the first Add.
                // Need to model this is as a write to race with the read in Add.
                // As a consequence, can do the write only for the first waiter,
                // otherwise concurrent Waits will race with each other.
                race.Write(unsafe.Pointer(&wg.sema))
            }
            runtime_Semacquire(&wg.sema)
            if wg.state.Load() != 0 {
                panic("sync: WaitGroup is reused before previous Wait has returned")
            }
            if race.Enabled {
                race.Enable()
                race.Acquire(unsafe.Pointer(wg))
            }
            return
        }
    }
}

Add a new 评论

Some HTML is okay.