使用sync包中的Waitgroup结构,官方文档对它的描述：

1
2
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.

我们可以像下面这样使用waitgroup:

1
2
3
4
创建一个Waitgroup的实例，我们叫它wg
在每个goroutine启动前或者在里面调用wg.Add(1)，也可以在创建n个goroutine前调用wg.Add(n)
当每个goroutine完成任务后，调用wg.Done()
在等待所有goroutine的地方调用wg.Wait()，wg.Done()前所有goroutine会阻塞，当所有goroutine都调用完wg.Done()之后它会返回。

代码目录结构如下：

1
2
3
4
5
src
|---main
    |---main.go
|---mqmgr
    |---mqmgr.go

代码如下：

mqmgr.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
package mqmgr

import (
    "fmt"
    "sync"
)

type MQMgr struct {
    addr      string
    port      int
    userName  string
    userPwd   string
    chanList  chan int
    waitGroup *sync.WaitGroup
}

func NewRabbitMQ() *MQMgr {
    return &MQMgr{
        chanList:  make(chan int, 1024),
        waitGroup: &sync.WaitGroup{},
    }
}

// 启动
func (m *MQMgr) Start() {
    m.waitGroup.Add(2)
    for i := 0; i < 2; i++ {
        go m.consumer()
    }
}

// 停止
func (m *MQMgr) Stop() {
    close(m.chanList)
    m.waitGroup.Wait()
}

// 向队列添加消息
func (m *MQMgr) PushMsg(val int) {
    m.chanList <- val
}

// 消费
func (m *MQMgr) consumer() {
    defer m.waitGroup.Done()
    for {
        select {
        case msg, ok := <-m.chanList:
            if ok {
                fmt.Println(msg)
            } else {
                fmt.Println("exit consumer")
                return
            }
        }
    }
}

main.go

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
package main

import (
    "fmt"
    "mqmgr"
    "os"
    "os/signal"
    "syscall"
)

func main() {
    mqDemo := mqmgr.NewRabbitMQ()
    mqDemo.Start()

    // Handle SIGINT and SIGTERM.
    ch := make(chan os.Signal)
    signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
    fmt.Println(<-ch)

    // Stop the service gracefully.
    mqDemo.Stop()
}