This table shows how processing time scales as we increase the number of consumer workers.
Each worker processes messages with a simulated delay of 100ms per message.
| Workers | Messages | Delay (ms) | Duration |
|---|---|---|---|
| 5 | 1000 | 100 | 20.234953s |
| 10 | 1000 | 100 | 10.114531s |
| 20 | 1000 | 100 | 5.060038208s |
| 40 | 1000 | 100 | 2.539948333s |
| 80 | 1000 | 100 | 1.315284208s |
| 160 | 1000 | 100 | 713.914791ms |
| 320 | 1000 | 100 | 413.064875ms |
| 640 | 1000 | 100 | 220.453167ms |
Observation: Scaling workers has a huge impact on throughput
| services: | |
| rabbitmq: | |
| image: rabbitmq:3-management | |
| container_name: rabbitmq | |
| ports: | |
| - "5672:5672" | |
| - "15672:15672" | |
| environment: | |
| RABBITMQ_DEFAULT_USER: guest | |
| RABBITMQ_DEFAULT_PASS: guest | |
| healthcheck: | |
| test: [ "CMD", "rabbitmq-diagnostics", "ping" ] | |
| interval: 5s | |
| timeout: 5s | |
| retries: 10 | |
| producer: | |
| build: | |
| context: . | |
| dockerfile: Dockerfile.producer | |
| depends_on: | |
| rabbitmq: | |
| condition: service_healthy | |
| env_file: | |
| - common.env | |
| entrypoint: [ "/bin/sh", "-c", "sleep 5 && /app/producer" ] | |
| restart: "no" | |
| consumer: | |
| build: | |
| context: . | |
| dockerfile: Dockerfile.consumer | |
| depends_on: | |
| rabbitmq: | |
| condition: service_healthy | |
| env_file: | |
| - common.env | |
| - consumer.env | |
| entrypoint: [ "/bin/sh", "-c", "sleep 10 && /app/consumer" ] | |
| restart: "no" |
| # ---- Builder Stage ---- | |
| FROM golang:1.23-alpine AS builder | |
| WORKDIR /app | |
| COPY go.mod go.sum ./ | |
| RUN go mod download | |
| COPY . . | |
| # Build the consumer binary | |
| RUN go build -o consumer ./cmd/consumer | |
| # ---- Runtime Stage ---- | |
| FROM alpine:latest | |
| # Copy the binary from builder | |
| COPY --from=builder /app/consumer /app/consumer | |
| # Run the consumer | |
| ENTRYPOINT ["/app/consumer"] |
| ... | |
| // Internal Go channel between your message consumer and your worker goroutines | |
| jobChan := make(chan amqp.Delivery) | |
| var wg sync.WaitGroup | |
| // Launch workers | |
| for i := range numWorkers { | |
| wg.Add(1) | |
| go func(id int) { | |
| defer wg.Done() | |
| worker(id, delayMs, jobChan) // workers will exit when they finish their current job | |
| }(i) | |
| } | |
| log.Println("Waiting for messages...") | |
| processedMessages := 0 | |
| for msg := range msgs { | |
| jobChan <- msg | |
| processedMessages++ | |
| if processedMessages >= numMessages { | |
| log.Printf("Processed %d messages, stopping consumer", processedMessages) | |
| close(jobChan) // Close the job channel to signal workers to stop | |
| break | |
| } | |
| } | |
| wg.Wait() // Wait for all workers to finish | |
| } | |
| func worker(id, delayMs int, jobs <-chan amqp.Delivery) { | |
| for job := range jobs { | |
| log.Printf("Worker %d processing message: %s", id, job.Body) | |
| time.Sleep(time.Duration(delayMs) * time.Millisecond) // Simulate work | |
| log.Printf("Worker %d done processing message: %s", id, job.Body) | |
| } | |
| } |