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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"sync"
	"sync/atomic"
	"time"

	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
	"github.com/aws/aws-sdk-go/aws"
)

const (
	queueURL = "https://sqs.us-east-1.amazonaws.com/181560427716/nutrition-mock"
	// totalMessages = 100_000
	totalMessages = 10_000
	batchSize     = 10
	workerCount   = 20
)

func main() {
	ctx := context.Background()
	cfg, err := config.LoadDefaultConfig(ctx)
	if err != nil {
		log.Fatalf("falha ao carregar AWS config: %v", err)
	}
	client := sqs.NewFromConfig(cfg)

	start := time.Now()

	var sentBatches int64
	var sentMessages int64
	var errBatches int64

	jobs := make(chan []types.SendMessageBatchRequestEntry, workerCount)

	var wg sync.WaitGroup

	for w := 0; w < workerCount; w++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for batch := range jobs {
				input := &sqs.SendMessageBatchInput{
					QueueUrl: aws.String(queueURL),
					Entries:  batch,
				}
				resp, err := client.SendMessageBatch(ctx, input)
				if err != nil || len(resp.Failed) > 0 {
					atomic.AddInt64(&errBatches, 1)
					log.Printf("[worker %d] erro em batch: %v – falhas: %d", id, err, len(resp.Failed))
				} else {
					atomic.AddInt64(&sentBatches, 1)
					atomic.AddInt64(&sentMessages, int64(len(batch)))
					if sentBatches%100 == 0 {
						log.Printf("✔ %d batches enviados (%d mensagens)", sentBatches, sentMessages)
					}
				}
			}
		}(w)
	}

	go func() {
		defer close(jobs)
		var batch []types.SendMessageBatchRequestEntry
		for i := 1; i <= totalMessages; i++ {
			msg := types.SendMessageBatchRequestEntry{
				Id:          aws.String(fmt.Sprintf("msg-%d", i)),
				MessageBody: aws.String(fmt.Sprintf(`{"index":%d}`, i)),
			}
			batch = append(batch, msg)
			if len(batch) == batchSize {
				jobs <- batch
				batch = nil
			}
		}
		if len(batch) > 0 {
			jobs <- batch
		}
	}()

	wg.Wait()

	elapsed := time.Since(start)
	log.Println("================================================")
	log.Printf("Mensagens enviadas: %d", sentMessages)
	log.Printf("Batches enviados:   %d", sentBatches)
	log.Printf("Batches falhados:   %d", errBatches)
	log.Printf("Tempo total:        %s", elapsed)
	log.Println("================================================")

	os.Exit(0)

}

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
58
59
60
61
62
63
64
65
66
67
68
package main

import (
	"context"
	"log"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
)

const (
	queueURL       = "https://sqs.us-east-1.amazonaws.com/181560427716/nutrition-mock"
	totalMessages  = 1_000
	progressPeriod = 100 // log a cada 100 mensagens
)

func main() {
	ctx := context.Background()

	// Carrega credenciais e região
	cfg, err := config.LoadDefaultConfig(ctx)
	if err != nil {
		log.Fatalf("falha ao carregar config AWS: %v", err)
	}
	client := sqs.NewFromConfig(cfg)

	var count int
	start := time.Now()

	for count < totalMessages {

		resp, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
			QueueUrl:            aws.String(queueURL),
			MaxNumberOfMessages: 1,
			WaitTimeSeconds:     10,
		})
		if err != nil {
			log.Printf("erro no ReceiveMessage: %v", err)
			continue
		}
		if len(resp.Messages) == 0 {
			continue
		}

		_, err = client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
			QueueUrl:      aws.String(queueURL),
			ReceiptHandle: resp.Messages[0].ReceiptHandle,
		})
		if err != nil {
			log.Printf("falha ao deletar msg %s: %v", aws.ToString(resp.Messages[0].MessageId), err)
		}

		// Incrementa o contador de controle
		count++

		if count%progressPeriod == 0 {
			log.Printf("progresso: %d/%d msgs", count, totalMessages)
		}
	}

	elapsed := time.Since(start)
	log.Printf("Total de mensagens: %d", count)
	log.Printf("Tempo total: %s", elapsed)
	log.Printf("TPS Médio: %d", int(count)/int(elapsed.Seconds()))
}

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
// ...

	for count < totalMessages {

		resp, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
			QueueUrl:            aws.String(queueURL),
			MaxNumberOfMessages: 10, // Batch Size de 10 mensagens
			WaitTimeSeconds:     10,
		})
		if err != nil {
			log.Printf("erro no ReceiveMessage: %v", err)
			continue
		}
		if len(resp.Messages) == 0 {
			continue
		}

		for _, message := range resp.Messages {

			_, err = client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
				QueueUrl:      aws.String(queueURL),
				ReceiptHandle: message.ReceiptHandle,
			})
			if err != nil {
				log.Printf("falha ao deletar msg %s: %v", aws.ToString(message.MessageId), err)
			}

			// Incrementa o contador de controle
			count++

			if count%progressPeriod == 0 {
				log.Printf("progresso: %d/%d msgs", count, totalMessages)
			}

		}

	}

// ...

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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
package main

import (
	"context"
	"log"
	"sync"
	"sync/atomic"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
)

const (
	queueURL       = "https://sqs.us-east-1.amazonaws.com/181560427716/nutrition-mock"
	totalMessages  = 1_000
	batchSize      = 10
	workerCount    = 10
	progressPeriod = 100 // log a cada 100 mensagens
)

func main() {
	ctx := context.Background()
	cfg, err := config.LoadDefaultConfig(ctx)
	if err != nil {
		log.Fatalf("falha ao carregar config AWS: %v", err)
	}
	client := sqs.NewFromConfig(cfg)

	var processed int64
	var wg sync.WaitGroup
	start := time.Now()

	// lança os workers
	for i := 1; i <= workerCount; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for {
				// Checa se o total de mensagens processadas já atingiu o limite - POC
				cur := atomic.LoadInt64(&processed)
				if cur >= totalMessages {
					return
				}

				resp, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
					QueueUrl:            aws.String(queueURL),
					MaxNumberOfMessages: batchSize,
					WaitTimeSeconds:     1,
					VisibilityTimeout:   30,
				})
				if err != nil {
					log.Printf("[worker %d] ReceiveMessage erro: %v", id, err)
					continue
				}
				if len(resp.Messages) == 0 {
					continue
				}

				for _, msg := range resp.Messages {
					_, err := client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
						QueueUrl:      aws.String(queueURL),
						ReceiptHandle: msg.ReceiptHandle,
					})
					if err != nil {
						log.Printf("[worker %d] falha ao deletar msg %s: %v", id, aws.ToString(msg.MessageId), err)
						continue
					}

					// incrementa contador
					newCount := atomic.AddInt64(&processed, 1)
					if newCount%progressPeriod == 0 {
						log.Printf("progresso: %d/%d mensagens", newCount, totalMessages)
					}
					if newCount >= totalMessages {
						return
					}
				}
			}
		}(i)
	}

	// espera os workers terminarem
	wg.Wait()
	elapsed := time.Since(start)
	log.Printf("Total de mensagens: %d", processed)
	log.Printf("Tempo total: %s", elapsed)
	log.Printf("TPS Médio: %d", int(processed)/int(elapsed.Seconds()))
}

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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
package main

import (
	"context"
	"log"
	"sync"
	"sync/atomic"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
)

const (
	queueURL       = "https://sqs.us-east-1.amazonaws.com/181560427716/nutrition-mock"
	totalMessages  = 100_000 // meta de consumo
	batchSize      = 10      // até 10 mensagens por ReceiveMessage
	workerCount    = 10      // paralelismo
	progressPeriod = 1000    // log a cada 100 mensagens
)

func main() {
	ctx := context.Background()
	cfg, err := config.LoadDefaultConfig(ctx)
	if err != nil {
		log.Fatalf("falha ao carregar AWS config: %v", err)
	}
	client := sqs.NewFromConfig(cfg)

	// canal de batches e sinal de término
	batchCh := make(chan []types.Message)
	doneCh := make(chan struct{})
	var once sync.Once
	signalDone := func() { once.Do(func() { close(doneCh) }) }

	var processed int64
	start := time.Now()

	var fetchWg, deleteWg sync.WaitGroup

	for i := 1; i <= workerCount; i++ {
		fetchWg.Add(1)
		go func(id int) {
			defer fetchWg.Done()
			for {
				// Trava do teste
				if atomic.LoadInt64(&processed) >= totalMessages {
					return
				}
				resp, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
					QueueUrl:            aws.String(queueURL),
					MaxNumberOfMessages: batchSize,
					WaitTimeSeconds:     1,
					VisibilityTimeout:   60,
				})
				if err != nil {
					log.Printf("[%d] Receive erro: %v", id, err)
					time.Sleep(time.Second)
					continue
				}
				if len(resp.Messages) == 0 {
					continue
				}
				// envia batch para o channel de processamento
				select {
				case batchCh <- resp.Messages:
				case <-doneCh:
					return
				}
			}
		}(i)
	}

	// quando todos os fetchers terminarem, fechamos batchCh
	go func() {
		fetchWg.Wait()
		close(batchCh)
	}()

	// Inicia os Workers de Processamento
	for i := 1; i <= workerCount; i++ {
		deleteWg.Add(1)
		go func(id int) {
			defer deleteWg.Done()
            // Consome as mensagens do SQS através do Channel de forma assincrona
			for batch := range batchCh {

				entries := make([]types.DeleteMessageBatchRequestEntry, 0, len(batch))
				for _, msg := range batch {
					entries = append(entries, types.DeleteMessageBatchRequestEntry{
						Id:            msg.MessageId,
						ReceiptHandle: msg.ReceiptHandle,
					})
				}

				if _, err := client.DeleteMessageBatch(ctx, &sqs.DeleteMessageBatchInput{
					QueueUrl: aws.String(queueURL),
					Entries:  entries,
				}); err != nil {
					log.Printf("[%d] DeleteBatch erro: %v", id, err)
				}

				newCount := atomic.AddInt64(&processed, int64(len(entries)))
				if newCount >= totalMessages {
					signalDone()
				}
				if newCount%progressPeriod == 0 || newCount >= totalMessages {
					log.Printf("progresso: %d/%d msgs", newCount, totalMessages)
				}
			}
		}(i)
	}

	deleteWg.Wait()

	elapsed := time.Since(start)
	final := atomic.LoadInt64(&processed)
	log.Printf("Total de mensagens: %d", final)
	log.Printf("Tempo total: %s", elapsed)
	log.Printf("TPS Médio: %d", int(final)/int(elapsed.Seconds()))

}