将多个 goroutine 与携带 amqp 消息的通道同步

当前位置： > > > > 将多个 goroutine 与携带 amqp 消息的通道同步

来源：stackoverflow
2024-04-21 19:27:33
0浏览
收藏

有志者，事竟成！如果你在学习Golang，那么本文《将多个 goroutine 与携带 amqp 消息的通道同步》，就很适合你！文章讲解的知识点主要包括，若是你对本文感兴趣，或者是想搞懂其中某个知识点，就请你继续往下看吧~

问题内容

我有点不明白为什么下面的设计会造成内存泄漏。能够通过简单的 kubectl top pod 命令重现泄漏，并在执行几秒钟后观察内存达到顶峰。

我有一个工作人员监视 rabbitmq 消息队列中的消息。如果我将约 100k 条小消息（每条约 200b）放入交换中，我会看到以下 go 应用程序开始泄漏。

在重写之前，我有时间和精力深入研究并了解到底是什么导致了内存泄漏。

type workerMessage struct {
    AmqpMessage amqp.Delivery
    Data        types.Data
}


func main() {
    var wg sync.WaitGroup
    for i := 0; i < opts.WorkerCount; i++ {
        wg.Add(1)
        go w.StartWork(&wg)
    }

    // create a wait group with 1 routine added
    var wait sync.WaitGroup
    wait.Add(1)

    // capture 2 signals, and have them send to signalChan
    stopChan := make(chan os.Signal, 1)
    signal.Notify(stopChan, syscall.SIGINT, syscall.SIGTERM, os.Kill, os.Interrupt)

    // wait for a receive on the signal channel, and decrement the wait group
    // when a singal is received
    go func() {
        <-stopChan
        wait.Done()
    }()

    // block until a signal is received
    wait.Wait()

}

func (w *Worker) StartWork(wg *sync.WaitGroup) {
    defer wg.Done()


    // messages will be acked from AMQP here
    chToBeAcked := make(chan workerMessage)
    go w.watchForAMQPMessagesToAck(&iwg, chToBeAcked)

    // data will be written to cassandra or object storage
    chDataToStorage := make(chan workerMessage)
    go w.writeDataToStorage(&iwg, chDataToStorage, chToBeAcked)

    // call services, process data and decides what to do next
    chProcessData := make(chan workerMessage)
    go w.ProcessData(&iwg, chProcessData, chDataToStorage, chToBeAcked)

    messageChannel, err := amqpChannel.Consume(
        queue.Name,
        ....
    )

    stopChan := make(chan bool)
    go func() {

        for message := range messageChannel {
            wm := workerMessage{
                AmqpMessage: message,
            }
            chProcessData <- wm
        }
    }()
    <-stopChan
}

func (w *Worker) ProcessData(
    wg *sync.WaitGroup,
    chProcessData <-chan workerMessage,
    chDataToStorage chan workerMessage,
    chMessagesWaitingForAck chan workerMessage,
    ){
    defer wg.Done()
    for msg := range chProcessData {
        ...
        make some API calls
        update some things in msg.Data
        ...

        If fastExit {
            chMessagesWaitingForAck <- msg
        } else {
            chDataToStorage <- msg
        }
    }
}

func (w *Worker) writeDataToStorage(
    wg *sync.WaitGroup,
    chDataToStorage <-chan workerMessage,
    chMessagesWaitingForAck chan workerMessage,
    ){
    defer wg.Done()
    for msg := range chDataToStorage {
        ...
        write data to cassandra or key/val object storage
        ...
        chMessagesWaitingForAck <- msg
    }
}

func (w *Worker) watchForAMQPMessagesToAck(
    wg *sync.WaitGroup,
    chMessagesWaitingForAck <-chan workerMessage,
){
    defer wg.Done()
    for msg := range chMessagesWaitingForAck {
        err := msg.AmqpMessage.Ack(false)
        w.stopOnError("failed to ack a message", msg, err)
    }
}

解决方案

问题是为每条消息创建 goroutine！纯粹是睡眠不足造成的疏忽

更新我的问题，这样就没有人会犯同样的错误了

本篇关于《将多个 goroutine 与携带 amqp 消息的通道同步》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于Golang的相关知识，请关注公众号！