Use context.Context to cancel propagation to ensure that the sub-goroutines can be terminated in time; 2. Use errgroup.Group to achieve error aggregation and rapid failure of concurrent tasks; 3. Use pipeline mode to improve data processing throughput through fan-out-fan-in; 4. Use sync.Once and atomic.Value to achieve efficient concurrent initialization; 5. Use rate.Limiter to control the request rate to prevent system overload; 6. Avoid memory leaks in time. After in select, use stopable timer instead; 7. Use atomic.Value to achieve lock-free configuration hot updates, requiring storage of immutable objects; these modes need to be combined, using context as the life cycle basis, and combining error processing, current limiting and atomic operations to build a robust and efficient concurrent system.

Go's concurrency model is known for its simplicity and efficiency, goroutine and channel at its core. However, with the increase in system complexity, using basic go and chan alone is no longer enough to cope with complex concurrent control needs. Mastering advanced concurrency mode can help us write more robust, maintainable, and high-performance concurrency programs.

Here are a few advanced Go Concurrency Patterns that are very useful in actual development, combined with usage scenarios and best practices.

1. Context and Cancellation Propagation

In a concurrent program, an operation may start multiple subtasks (goroutines). When the main task is cancelled, all subtasks should be terminated in time to avoid wasting resources.

Key point: Use context.Context to manage life cycles uniformly.

 ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()

resultCh := make(chan string, 1)
go func() {
    result := slowOperation(ctx) // ctx.Done() is required to be listened to inside the function
    select {
    case resultCh <- result:
    default:
    }
}()

select {
case <-ctx.Done():
    log.Println("operation cancelled:", ctx.Err())
case result := <-resultCh:
    log.Println("got result:", result)
}

?Best Practice :

• All long-running functions should accept context.Context parameter.
• Check ctx.Done() regularly in the for loop, or listen with select .
• Use context.WithCancel , WithTimeout , WithDeadline to build a hierarchical cancel tree.

2. ErrGroup: Error aggregation of concurrent tasks

When multiple tasks need to be executed concurrently and you want to fail quickly when any task errors occur, while waiting for all tasks to be cleaned up, errgroup.Group is the best choice.

 g, ctx := errgroup.WithContext(context.Background())

urls := []string{"http://example1.com", "http://example2.com"}

for _, url := range urls {
    url := url
    g.Go(func() error {
        req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
        resp, err := http.DefaultClient.Do(req)
        if err != nil {
            return err
        }
        defer resp.Body.Close()
        // Process the response...
        return nil
    })
}

if err := g.Wait(); err != nil {
    log.Printf("failed to fetch: %v", err)
}

? Advantages :

• Automatically propagate context to all subtasks.
• As long as one task returns a non- nil error, other tasks are cancelled through ctx .
• Wait for all startup goroutines to complete (even if there is an error).

3. Pipeline mode: Fan-out/Fan-in

Split the data stream into multiple concurrent processing stages to improve throughput. Commonly found in data processing pipelines.

 func gen(ctx context.Context, nums ...int) <-chan int {
    out := make(chan int, len(nums))
    go func() {
        defer close(out)
        for _, n := range nums {
            select {
            case out <- n:
            case <-ctx.Done():
                Return
            }
        }
    }()
    return out
}

func sq(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case out <- n * n:
            case <-ctx.Done():
                Return
            }
        }
    }()
    return out
}

// Use ctx, cancel := context.WithCancel(context.Background())
defer cancel()

nums := gen(ctx, 1, 2, 3, 4)
squared := sq(ctx, nums)

for n := range squared {
    fmt.Println(n)
}

? Optimization Tips :

• Concurrent processing of multiple workers (fan-out):

   // Start multiple sq workers
  var chans []<-chan int
  for i := 0; i < 3; i {
      chans = append(chans, sq(ctx, nums))
  }
  // Merge results (fan-in)
  merged := merge(ctx, chans...)

• Use the merge function to merge multiple channels into one.

4. Dual check lock sync.Once (control initialization concurrency)

Although sync.Once can guarantee a single execution, in some scenarios, manually implementing dual-check locks can reduce lock competition.

 var once sync.Once
var client *http.Client

func GetClient() *http.Client {
    once.Do(func() {
        client = &http.Client{Timeout: 10 * time.Second}
    })
    Return client
}

??Note : Go's sync.Once has been optimized internally, and there is usually no need for manual double check locks. However, in extreme performance scenarios, lock-free reading can be achieved in combination with atomic.Value :

 var client atomic.Value

func initClient() {
    client.Store(&http.Client{Timeout: 10 * time.Second})
}

func GetClient() *http.Client {
    c := client.Load()
    if c == nil {
        once.Do(initClient)
        c = client.Load()
    }
    return c.(*http.Client)
}

5. Current limiter (Rate Limited) and resource control

Use golang.org/x/time/rate to achieve smooth current limit to prevent system overload.

 limiter := rate.NewLimiter(10, 1) // 10 per second, 1 burst for {
    if err := limiter.Wait(ctx); err != nil {
        break
    }
    go processRequest()
}

?Applicable scenarios :

• Call third-party API to limit current.
• Controls the number of concurrency of background tasks.
• Prevent database connections from skyrocketing.

6. Select timeout and default branch traps

select is the core control structure of Go concurrency, but it is prone to errors when used improperly.

 select {
case msg := <-ch:
    fmt.Println("received:", msg)
case <-time.After(1 * time.Second):
    fmt.Println("timeout")
}

?Problem : time.After creates a timer every time it is called, which can cause memory leaks (especially in loops).

?Fix : Use time.NewTimer and stop manually:

 timer := time.NewTimer(1 * time.Second)
select {
case msg := <-ch:
    fmt.Println(msg)
    if !timer.Stop() {
        <-timer.C // Empty channel
    }
case <-timer.C:
    fmt.Println("timeout")
}

7. Concurrent and secure configuration hot update: atomic.Value

Avoid using mutex to protect the entire configuration structure and use atomic.Value to implement lock-free reading.

 var config atomic.Value

// Initialize config.Store(loadConfig())

// Read (high frequency)
current := config.Load().(*Config)

// Update (low frequency)
newCfg := loadConfig()
config.Store(newCfg)

?Requirements : The stored objects must be immutable, or stored after copying.

Summarize

These advanced concurrency patterns are not isolated and often require combinations:

• Use context to control the life cycle.
• Use errgroup to manage concurrent tasks errors.
• Build data flow with pipeline .
• Use rate.Limiter to control resources.
• Use atomic.Value to achieve high-performance configuration updates.

Go's concurrency philosophy is: "Don't communicate through shared memory, but share memory through communication." But the advanced model tells us: Only by rationally using locks, atomic operations and context control between performance and control can we build a truly reliable system.

Basically all this is not complicated but easy to ignore.

The above is the detailed content of Advanced Go Concurrency Patterns. For more information, please follow other related articles on the PHP Chinese website!