RabbitMQ连接断开时，如何关闭所有监听notifyClose的goroutine？

问题描述 #

我编写了如下connectingBalancing函数，初始化连接和channel时会循环调用该函数。现在遇到的问题是：当其中一个连接断开时，该如何终止所有相关的goroutine？我曾考虑使用signal.Notify()，但不知具体如何实现。

func (c Connects) connectingBalancing(
    conn connect,
    channel *amqp.Channel,
    consumer Consumer,
) {
    type chanErr chan *amqp.Error
    var notifyConnClose chanErr

    if conn.err != nil {
        notifyConnClose = conn.err
    } else {
        notifyConnClose = conn.conn.NotifyClose(make(chanErr))
    }

    notifyChanClose := channel.NotifyClose(make(chanErr))

    for notifyConnClose != nil || notifyChanClose != nil {
        select {
        case err, ok := <-notifyConnClose:
            if !ok {
                notifyConnClose = nil
            } else {
                fmt.Println("connection closed, error", err)
            }
        case err, ok := <-notifyChanClose:
            if !ok {
                notifyChanClose = nil
            } else {
                fmt.Println("connection closed, error", err)
                channelStatus = false

                time.Sleep(time.Second * 1)

                newCn, err := conn.conn.Channel()
                if err != nil {
                    log.Println(err)
                }

                if err := c.createChannel(newCn, consumer); err != nil {
                    log.Println(err)
                }

                channel = newCn
                notifyChanClose = channel.NotifyClose(make(chanErr))
            }
        }
    }
}

解决方案 #

核心思路：用共享退出通道统一管理终止信号 #

signal.Notify()主要用于捕获系统信号（如Ctrl+C），但针对连接断开时主动终止所有相关goroutine的需求，更适合通过一个共享的退出通道来广播终止指令，结合AMQP原生的连接/channel关闭通知一起监听。

具体实现步骤 #

1. 给Connects结构体添加共享退出通道：用于在连接断开时向所有相关goroutine发送终止信号。
2. 在connectingBalancing中监听退出信号：将退出通道加入select逻辑，一旦收到信号就直接退出循环、终止当前goroutine。
3. 连接断开时触发全局终止：捕获到连接关闭错误时，关闭退出通道（通道关闭后所有监听它的goroutine都会收到退出信号）。

修改后的代码示例 #

// 给Connects结构体添加quit通道，用于广播终止信号
type Connects struct {
    // 保留原有字段...
    quit chan struct{}
}

// 初始化Connects时创建quit通道
func NewConnects() *Connects {
    return &Connects{
        quit: make(chan struct{}),
    }
}

func (c Connects) connectingBalancing(
    conn connect,
    channel *amqp.Channel,
    consumer Consumer,
) {
    type chanErr chan *amqp.Error
    var notifyConnClose chanErr

    if conn.err != nil {
        notifyConnClose = conn.err
    } else {
        notifyConnClose = conn.conn.NotifyClose(make(chanErr))
    }

    notifyChanClose := channel.NotifyClose(make(chanErr))

    for {
        select {
        // 监听全局终止信号
        case <-c.quit:
            fmt.Println("收到终止信号，退出当前goroutine")
            // 执行清理操作：关闭当前channel和连接
            if channel != nil {
                _ = channel.Close()
            }
            if conn.conn != nil {
                _ = conn.conn.Close()
            }
            return
        case err, ok := <-notifyConnClose:
            if !ok {
                notifyConnClose = nil
            } else {
                fmt.Println("connection closed, error", err)
                // 连接断开，触发全局终止
                close(c.quit)
            }
        case err, ok := <-notifyChanClose:
            if !ok {
                notifyChanClose = nil
            } else {
                fmt.Println("channel closed, error", err)
                channelStatus = false

                time.Sleep(time.Second * 1)

                newCn, err := conn.conn.Channel()
                if err != nil {
                    log.Println(err)
                    // 重建channel失败，触发全局终止
                    close(c.quit)
                    return
                }

                if err := c.createChannel(newCn, consumer); err != nil {
                    log.Println(err)
                    close(c.quit)
                    return
                }

                channel = newCn
                notifyChanClose = channel.NotifyClose(make(chanErr))
            }
        }
    }
}

补充：结合系统信号实现手动终止 #

如果需要支持用户通过系统信号（如Ctrl+C）终止所有goroutine，可以在主函数中添加以下逻辑：

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

func main() {
    connects := NewConnects()

    // 捕获系统中断信号
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

    go func() {
        <-sigChan
        fmt.Println("收到系统中断信号，终止所有goroutine")
        close(connects.quit)
    }()

    // 初始化连接、启动goroutine...
}

注意：close(c.quit)只能调用一次，多次关闭会触发panic，需确保仅在真正需要全局终止时调用。

内容的提问来源于stack exchange，提问作者blackmarllbor0