The answer is to implement Worker Pool mode using channel and goroutine to control the number of concurrent and efficiently process batch tasks. 1. Define the task job and result Result structure; 2. Create buffered task channel and result channel; 3. Start a fixed number of worker goroutines, each worker reads the task from the jobs channel and writes the results channel, and calls wg.Done() after completion; 4. The main goroutine sends all tasks and closes the jobs channel; 5. Use sync.WaitGroup to wait for all workers to complete, and closes the results channel in the goroutine; 6. The main goroutine collects results from results to count the number of successes and failures. This mode prevents resource exhaustion by limiting the number of goroutines, is suitable for batch task processing, and can be extended and optimized by adding context timeouts, retry mechanisms, etc. The final program correctly outputs the processing results and exits safely.

In Go, Worker Pool is a common concurrency pattern that is used to efficiently handle large numbers of tasks while limiting the number of goroutines running simultaneously and avoiding resource exhaustion. It is very suitable for handling batch tasks such as file processing, network requests, data cleaning, etc.

The following is a specific example to demonstrate how to implement a simple Worker Pool using Go.

?? Scenario: Process a batch of tasks (such as downloading URLs)

Suppose we have 100 tasks (such as HTTP requests), but we only want to run 10 goroutines at the same time to handle these tasks to prevent system overload.

? Core structure

• A jobs channel
• Multiple workers (goroutines)
• A result queue ( results channel)
• Wait for all tasks to complete ( sync.WaitGroup )

? Complete sample code

 package main

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

// Task structure type Job struct {
    ID int
}

// Result structure type Result struct {
    JobID int
    Success bool
    Message string
}

func worker(id int, jobs <-chan Job, results chan<- Results, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("Worker %d started job %d\n", id, job.ID)

        // Simulate time-consuming operations (such as network requests)
        time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)

        success := rand.Float32() > 0.3 // 70% success rate var msg string
        if success {
            msg = "processed successfully"
        } else {
            msg = "failed"
        }

        results <- Results{
            JobID: job.ID,
            Success: success,
            Message: msg,
        }
        fmt.Printf("Worker %d finished job %d\n", id, job.ID)
    }
}

func main() {
    const numJobs = 100
    const numWorkers = 10

    jobs := make(chan Job, numJobs)
    results := make(chan Results, numJobs)

    var wg sync.WaitGroup

    // 1. Start workers
    for i := 1; i <= numWorkers; i {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // 2. Send task for i := 1; i <= numJobs; i {
        jobs <- Job{ID: i}
    }
    close(jobs) // Close jobs channel, indicating that there are no more tasks // 3. Wait for all workers to complete in another goroutine, and then close results
    go func() {
        wg.Wait()
        close(results)
    }()

    // 4. Collect results successfully := 0
    failed := 0
    for result := range results {
        if result.Success {
            Successful  
        } else {
            failed  
        }
        // Optional: Print or record the result // fmt.Printf("Job %d: %s\n", result.JobID, result.Message)
    }

    fmt.Printf("Processing complete. Success: %d, Failed: %d\n", successful, failed)
}

? Key points description

? jobs and results use buffered channel

• jobs := make(chan Job, numJobs) : Buffer the channel to avoid send blockage
• It can also not buffer, but when sending tasks, make sure that a worker is receiving it.

?After close(jobs) , range jobs will automatically exit

• All workers' for job := range jobs will end naturally after the channel is closed

? Use sync.WaitGroup to wait for all workers to exit

• Each worker completes wg.Done()
• The main goroutine closes results after waiting for completion to prevent range results from deadlocking

? results need to be closed as well

• Otherwise for range results will block forever

? Scalable optimization

• Error retry mechanism : Add retry logic to the worker
• Timeout control : Use context.WithTimeout
• Dynamic worker number : adjust according to load
• Priority task queue : Use multiple channels or priority queues

? Applicable scenarios

• Batch processing of files or database records
• Concurrent crawling of web pages
• Email Send Queue
• Image/video transcoding
• Log processing

Basically that's it. Go's channel goroutine makes the worker pool implementation very simple and intuitive. The key is to understand the use of channel shutdown and WaitGroup to avoid deadlocks or leaks.

The above is the detailed content of go by example worker pools. For more information, please follow other related articles on the PHP Chinese website!