Я немного растерялся, потому что следующий дизайн создает утечку памяти. Способен воспроизвести утечку с помощью простой команды kubectl top pod и наблюдает, как память падает на крышу через несколько секунд выполнения.
У меня есть работник, который следит за очередью сообщений RabbitMQ для сообщений. Если я добавлю ~ 100К небольших сообщений (каждое ~ 200В) в обмен, я вижу, что следующее 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)
}
}