我有点不明白为什么下面的设计会造成内存泄漏。能够通过一个简单的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)
}
}
问题是为每条消息创建goroutines！由于缺乏睡眠而造成的纯粹疏忽
更新我的问题，这样就没有人会犯同样的错误