Concurrency паттерны в Go

1func fanOut(input <-chan int, workers int) []<-chan int {
2    channels := make([]<-chan int, workers)
3    for i := 0; i < workers; i++ {
4        channels[i] = process(input)
5    }
6    return channels
7}
8
9func process(in <-chan int) <-chan int {
10    out := make(chan int)
11    go func() {
12        defer close(out)
13        for n := range in {
14            out <- n * n
15        }
16    }()
17    return out
18}

1func fanIn(channels ...<-chan int) <-chan int {
2    var wg sync.WaitGroup
3    merged := make(chan int)
4
5    for _, ch := range channels {
6        wg.Add(1)
7        go func(c <-chan int) {
8            defer wg.Done()
9            for value := range c {
10                merged <- value
11            }
12        }(ch)
13    }
14
15    go func() {
16        wg.Wait()
17        close(merged)
18    }()
19
20    return merged
21}

1func main() {
2    input := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
3
4    // Fan-out на 3 воркера
5    workers := fanOut(input, 3)
6
7    // Fan-in
8    for result := range fanIn(workers...) {
9        fmt.Println(result)
10    }
11}

1func generate(nums ...int) <-chan int {
2    out := make(chan int)
3    go func() {
4        for _, n := range nums {
5            out <- n
6        }
7        close(out)
8    }()
9    return out
10}
11
12func square(in <-chan int) <-chan int {
13    out := make(chan int)
14    go func() {
15        for n := range in {
16            out <- n * n
17        }
18        close(out)
19    }()
20    return out
21}
22
23func filter(in <-chan int, predicate func(int) bool) <-chan int {
24    out := make(chan int)
25    go func() {
26        for n := range in {
27            if predicate(n) {
28                out <- n
29            }
30        }
31        close(out)
32    }()
33    return out
34}
35
36func main() {
37    // Pipeline: generate -> filter -> square
38    ch := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
39    ch = filter(ch, func(n int) bool { return n%2 == 0 })
40    ch = square(ch)
41
42    for result := range ch {
43        fmt.Println(result) // 4, 16, 36, 64, 100
44    }
45}

1func workerPool(jobs <-chan Job, results chan<- Result, numWorkers int) {
2    var wg sync.WaitGroup
3
4    for i := 0; i < numWorkers; i++ {
5        wg.Add(1)
6        go func(workerID int) {
7            defer wg.Done()
8            for job := range jobs {
9                result := process(job)
10                results <- result
11            }
12        }(i)
13    }
14
15    go func() {
16        wg.Wait()
17        close(results)
18    }()
19}

1func main() {
2    jobs := make(chan Job, 100)
3    results := make(chan Result, 100)
4
5    // Запуск воркеров
6    workerPool(jobs, results, 5)
7
8    // Отправка задач
9    for i := 0; i < 20; i++ {
10        jobs <- Job{ID: i}
11    }
12    close(jobs)
13
14    // Сбор результатов
15    for result := range results {
16        fmt.Println(result)
17    }
18}

1func main() {
2    requests := make(chan int, 5)
3    for i := 1; i <= 5; i++ {
4        requests <- i
5    }
6    close(requests)
7
8    // Ограничитель: 2 запроса в секунду
9    limiter := time.NewTicker(200 * time.Millisecond)
10    defer limiter.Stop()
11
12    for req := range requests {
13        <-limiter.C
14        fmt.Println("request", req, time.Now())
15    }
16}

1type RateLimiter struct {
2    ticker *time.Ticker
3    quit   chan struct{}
4}
5
6func NewRateLimiter(rps int) *RateLimiter {
7    return &RateLimiter{
8        ticker: time.NewTicker(time.Second / time.Duration(rps)),
9        quit:   make(chan struct{}),
10    }
11}
12
13func (rl *RateLimiter) Wait() {
14    <-rl.ticker.C
15}
16
17func (rl *RateLimiter) Stop() {
18    rl.ticker.Stop()
19    close(rl.quit)
20}

1func longOperation(ctx context.Context) error {
2    select {
3    case <-time.After(5 * time.Second):
4        return nil
5    case <-ctx.Done():
6        return ctx.Err()
7    }
8}
9
10func main() {
11    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
12    defer cancel()
13
14    err := longOperation(ctx)
15    if err != nil {
16        fmt.Println("Operation cancelled:", err)
17    }
18}

1type contextKey string
2
3const userIDKey contextKey = "userID"
4
5func WithUserID(ctx context.Context, userID int) context.Context {
6    return context.WithValue(ctx, userIDKey, userID)
7}
8
9func GetUserID(ctx context.Context) (int, bool) {
10    userID, ok := ctx.Value(userIDKey).(int)
11    return userID, ok
12}

1import "runtime"
2
3func monitorGoroutines() {
4    ticker := time.NewTicker(time.Second)
5    for range ticker.C {
6        fmt.Printf("Goroutines: %d\n", runtime.NumGoroutine())
7    }
8}
9
10func main() {
11    go monitorGoroutines()
12    // Основная логика
13}

1func processWithTimeout(data []int) []int {
2    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
3    defer cancel()
4
5    results := make([]int, len(data))
6
7    var wg sync.WaitGroup
8    for i, v := range data {
9        wg.Add(1)
10        go func(i, v int) {
11            defer wg.Done()
12            select {
13            case <-ctx.Done():
14                return
15            default:
16                results[i] = expensiveCalculation(v)
17            }
18        }(i, v)
19    }
20
21    wg.Wait()
22    return results
23}

1func processAll(items []Item) {
2    var wg sync.WaitGroup
3    for _, item := range items {
4        wg.Add(1)
5        go func(item Item) {
6            defer wg.Done()
7            process(item)
8        }(item)
9    }
10    wg.Wait() // Гарантирует завершение всех горутин
11}