Golang The Series EP.170: Building an AI Batch Processing System with Golang Worker Pool

Welcome to EP.170! We have finally arrived at the grand finale workshop for the AI API Infrastructure chapter in the Superdev Academy series. After thoroughly learning about Multi-LLM management, Redis Cache, Rate Limiting, Load Balancing, Circuit Breaker, Prometheus Metrics, as well as Error Handling and Concurrency.

In the real enterprise world, AI workloads aren't just about replying to single chat messages (Transactional Queries). We often have to deal with massive "Batch Workloads", such as summarizing 10,000 customer feedback reviews, extracting table of contents data from thousands of PDF documents, or running weekly Sentiment Analysis.

If we send data one by one using a Synchronous Loop, the system could take days, and if a crash occurs midway, all the processed data might instantly vanish! Today, we will use Go to build a High-Performance AI Batch Processor that can process thousands of records in mere minutes, using a secure, stable, and fully monitored Worker Pool architecture!

AI Batch Processor System Architecture

Our system is divided into 4 main parts, working together via Goroutines and Channels:

Plaintext

+------------------+      +-------------------+      +-------------------+      +------------------+
|  Job Dispatcher  | ---> |    Worker Pool    | ---> |   AI Processing   | ---> |  Result Handler  |
| (Reads data &    |      | (Runs n parallel  |      |   (Calls API +    |      | (Saves results & |
| feeds Job Channel|      |      workers)     |      |  Circuit Breaker) |      |   Monitoring)    |
+------------------+      +-------------------+      +-------------------+      +------------------+

• Job Dispatcher: Distributes jobs into a Buffered Channel (JobQueue).

• Worker Pool: A specified number of Goroutines (e.g., 50 workers) pull jobs to process concurrently.

• AI Processing: Simulates/calls the LLM API with built-in timing and Error Tracking.

• Result Handler & Collector: A separate Goroutine that receives results from the ResultQueue to save into a Database/Log without interrupting the Workers.

Workshop Project Code: AI Batch Processing System

We will divide the code into 3 main parts for easier understanding. Create a main.go file and let's write it step-by-step.

Part 1: Data Structures (Structs & Initialization)

The first part is preparing the structures to store job data, results, and the Worker Pool controller.

Go

package main

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
)

// Job represents a single piece of work for the AI to process
type Job struct {
	ID      int
	Payload string
}

// Result represents the outcome of the processing (success or failure)
type Result struct {
	JobID  int
	Output string
	Err    error
}

// BatchProcessor controls the Worker Pool system
type BatchProcessor struct {
	WorkerCount  int
	JobQueue     chan Job
	ResultQueue  chan Result
	SuccessCount uint64
	FailureCount uint64
}

// NewBatchProcessor creates an instance and initializes queue sizes
func NewBatchProcessor(workerCount int, queueSize int) *BatchProcessor {
	return &BatchProcessor{
		WorkerCount: workerCount,
		JobQueue:    make(chan Job, queueSize),
		ResultQueue: make(chan Result, queueSize),
	}
}

Part 2: AI Simulation and Running the Worker Pool

The callAIModel function simulates calling an AI API (with simulated delays and random errors). The StartWorkers function creates the specified number of Goroutines to wait for jobs from the JobQueue.

Go

// callAIModel simulates sending data to be processed by an AI API
func (bp *BatchProcessor) callAIModel(ctx context.Context, job Job) (string, error) {
	// Simulate AI processing time (200ms - 600ms)
	processTime := time.Duration(200+rand.Intn(400)) * time.Millisecond

	select {
	case <-time.After(processTime):
		// Simulate a 5% random chance of failure
		if rand.Float32() < 0.05 {
			return "", fmt.Errorf("AI Provider Error on Job #%d", job.ID)
		}
		return fmt.Sprintf("Summary for Item %d: '%s' [Success]", job.ID, job.Payload), nil
	case <-ctx.Done():
		return "", ctx.Err()
	}
}

// StartWorkers spins up the specified number of Goroutine Workers
func (bp *BatchProcessor) StartWorkers(ctx context.Context, wg *sync.WaitGroup) {
	for i := 1; i <= bp.WorkerCount; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			for job := range bp.JobQueue {
				// Execute AI Processing
				output, err := bp.callAIModel(ctx, job)
				
				// Use atomic for thread-safe updates to counters (prevents Race Conditions)
				if err != nil {
					atomic.AddUint64(&bp.FailureCount, 1)
				} else {
					atomic.AddUint64(&bp.SuccessCount, 1)
				}
				
				// Send the result to the queue for handling
				bp.ResultQueue <- Result{JobID: job.ID, Output: output, Err: err}
			}
		}(i)
	}
}

Part 3: Main Function (Assembly and Execution)

In the final part, we will simulate feeding 1,000 jobs and collecting the results through the Collector to summarize the execution.

Go

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	totalJobs := 1000  // Total 1,000 jobs
	workerCount := 50  // Run 50 parallel workers

	processor := NewBatchProcessor(workerCount, totalJobs)

	var workerWg sync.WaitGroup
	var resultWg sync.WaitGroup

	startTime := time.Now()

	// 1. Start Worker Pool
	processor.StartWorkers(ctx, &workerWg)

	// 2. Result Collector Goroutine
	resultWg.Add(1)
	go func() {
		defer resultWg.Done()
		for result := range processor.ResultQueue {
			if result.Err != nil {
				// Real world: Save to Error Log or send to retry queue
			} else {
				// Real world: Save to Database or export as CSV/JSON
			}
		}
	}()

	// 3. Dispatcher: Feed 1,000 jobs into the Queue
	fmt.Printf("🚀 Starting to dispatch %d jobs into the Worker Pool (%d Workers)...\n", totalJobs, workerCount)
	for i := 1; i <= totalJobs; i++ {
		processor.JobQueue <- Job{
			ID:      i,
			Payload: fmt.Sprintf("Customer Feedback #%d", i),
		}
	}
	close(processor.JobQueue) // Close Job Channel to notify workers that there are no more jobs

	// 4. Wait for all Workers to finish, then close Result Queue
	workerWg.Wait()
	close(processor.ResultQueue)

	// 5. Wait for the Collector to finish gathering results
	resultWg.Wait()

	totalDuration := time.Since(startTime)

	// 6. Summarize the processing report
	fmt.Println("\n==============================================")
	fmt.Println("📊 AI Batch Processing Execution Summary")
	fmt.Println("==============================================")
	fmt.Printf("⏱️ Total Time Elapsed : %v\n", totalDuration)
	fmt.Printf("✅ Successful Jobs    : %d items\n", processor.SuccessCount)
	fmt.Printf("❌ Failed Jobs        : %d items\n", processor.FailureCount)
	fmt.Printf("⚡ Throughput (RPS)   : %.2f Requests/sec\n", float64(totalJobs)/totalDuration.Seconds())
	fmt.Println("==============================================")
}

Results and Performance Analysis

When you run the code above with go run main.go, you will see the following processing statistics:

Plaintext

🚀 Starting to dispatch 1000 jobs into the Worker Pool (50 Workers)...

==============================================
📊 AI Batch Processing Execution Summary
==============================================
⏱️ Total Time Elapsed : 8.12s
✅ Successful Jobs    : 952 items
❌ Failed Jobs        : 48 items
⚡ Throughput (RPS)   : 123.15 Requests/sec
==============================================

Why is this method so powerful?

• Reduces time by over 50x: If processed sequentially, 1,000 items × 400ms would take 400 seconds (~6.6 minutes). But with 50 Go Workers, we finish the processing in just 8 seconds!

• Memory Constrained: Using Bounded Channels and Atomic Counters helps control RAM usage from spiking, even when hundreds of thousands of jobs are queued up.

🎯 Daily Mission

Try running this Workshop code on your machine and tweak the workerCount (e.g., from 50 to 10 or 100) to compare the speed results.

Thought Experiment Homework: If during Batch Processing, a Rate Limit from the AI provider cuts you off (causing a 429 Too Many Requests error), how would you wrap the golang.org/x/time/rate package (which we learned in EP.164) around the workers' job-fetching phase to prevent exceeding the quota? Try tweaking the code!

🙋‍♂️ Frequently Asked Questions (FAQ)

Why use a Channel to distribute jobs instead of creating a slice and using a go func() loop directly?

Spawning an unlimited (Unbounded) number of Goroutines, like calling go func() 1,000 times concurrently, can lead to Out of Memory issues or Connection Exhaustion. Using a Worker Pool + Channel allows us to control Concurrency at an optimal level (like 50), making the system much more stable and reliable.

If I want the system to automatically "Retry" failed jobs, what should I do?

You can enhance the Result Collector. If it checks and finds result.Err != nil, instead of logging the error immediately, it can send result.JobID back into the JobQueue (or create a separate RetryQueue). But be careful to limit the maximum number of retries to prevent Infinite Loops.

What is the optimal number for workerCount?

For I/O Bound tasks like calling APIs or Database queries (like in this Workshop), the worker count can be set high (e.g., 50, 100, or 200), depending on the API provider's Rate Limit. But if it's a CPU Bound task (heavy computations on your machine), it should be set close to the number of CPU cores on your server.

Conclusion

In this article, we learned how to design and build an Enterprise-grade AI Batch Processing system using the Worker Pool architecture in Go, allowing us to handle massive workloads efficiently. We saw the power of Concurrency via Goroutines, safe data passing via Channels, and Asynchronous result handling. These are the core reasons why Go has become a highly popular language for backend systems.

Coming up next (EP.171): Congratulations! You have successfully completed the entire AI Backend Infrastructure curriculum. In the next episode, we will step into a "New Series / Advanced Topic": the world of AI Autonomous Agents. In "EP.171: Intro to AI Agents - When AI can decide to use tools on its own", we will explore how an LLM transforms from merely answering questions into a system that thinks, analyzes, runs code, or calls APIs to solve problems for us. Don't miss it, Gophers!

Follow Superdev Academy on all platforms:

• 🔵 Facebook: Superdev Academy Thailand

• 🎬 YouTube: Superdev Academy Channel

• 📸 Instagram: @superdevacademy

• 🎬 TikTok: @superdevacademy

• 🌐 Website: superdevacademy.com