主页 / coding / 问题 / 79241536
Accepted
onekram
Asked: 2024-12-01 20:25:18 +0800 CST

如何修复管道取消时发生的 goroutine 泄漏

  • 772

我使用线程池函数构建管道，并将 context.Context 作为参数传入其中。当调用 cancel() 函数或超时时，管道必须正常终止，这样就不会剩下任何可用的 goroutine。

我使用的职能：

func generate(amount int) <-chan int {
    result := make(chan int)
    go func() {
        defer close(result)
        for i := 0; i < amount; i++ {
            result <- i
        }
    }()

    return result
}

func sum(input <-chan int) int {
    result := 0
    for el := range input {
        result += el
    }
    return result
}

func process[T any, R any](ctx context.Context, workers int, input <-chan T, do func(T) R) <-chan R {
    wg := new(sync.WaitGroup)
    result := make(chan R)

    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case val, ok := <-input:
                    if !ok {
                        return
                    }
                    select {
                    case <-ctx.Done():
                        return
                    case result <- do(val):
                    }
                }
            }
        }()
    }

    go func() {
        defer close(result)
        wg.Wait()
    }()
    return result
}

用法：

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 1200*time.Millisecond)
    defer cancel()

    input := generate(1000)
    multiplied := process(ctx, 15, input, func(val int) int {
        time.Sleep(time.Second)
        return val * 2
    })
    increased := process(ctx, 15, multiplied, func(val int) int {
        return val + 10
    })

    fmt.Println("Result: ", sum(increased)) //  360 is ok
    fmt.Println("Num goroutine: ", runtime.NumGoroutine())  // 18 is too much
}

我理解发生这种情况是因为所有增加 goroutine 都结束了，而乘法 goroutine 仍在运行。

有没有什么规范的方法来解决这个问题？

multithreading

1 个回答

  1. Best Answer

    您期望类似结构化并发的东西，因此所有 goroutine 都应在当前范围的末尾结束，但不要按照您的期望设计代码。generate当输入通道未耗尽且您的do函数不可取消时，您将会泄漏。

    generate为你的功能添加可取消性do会有些帮助：

    package main

import (
    "context"
    "fmt"
    "runtime"
    "sync"
    "time"
)

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 1200*time.Millisecond)
    defer cancel()

    input := generate(ctx, 1_000)

    multiplied := process(ctx, 15, input, func(ctx context.Context, val int) (int, error) {
        select {
        case <-ctx.Done():
            return 0, ctx.Err()

        case <-time.After(time.Second):
            return val * 2, nil
        }
    })

    increased := process(ctx, 15, multiplied, func(_ context.Context, val int) (int, error) {
        return val + 10, nil
    })

    fmt.Println("Result: ", sum(increased))                //  360 is ok
    fmt.Println("Num goroutine: ", runtime.NumGoroutine()) // 18 is too much
}

func generate(ctx context.Context, amount int) <-chan int {
    input := make(chan int)
    go func() {
        defer close(input)
        for i := 0; i < amount; i++ {
            select {
            case <-ctx.Done():
                return

            case input <- i:
            }
        }
    }()

    return input
}

func sum(input <-chan int) int {
    result := 0
    for el := range input {
        result += el
    }
    return result
}

func process[T any, R any](ctx context.Context, workers int, input <-chan T, do func(context.Context, T) (R, error)) <-chan R {
    wg := new(sync.WaitGroup)
    result := make(chan R)

    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case val, ok := <-input:
                    if !ok {
                        return
                    }
                    r, err := do(ctx, val)
                    if err != nil {
                        return
                    }
                    result <- r
                }
            }
        }()
    }

    go func() {
        defer close(result)
        wg.Wait()
    }()

    return result
}

    “高级 Go 并发模式”中提到了更多内容，但作为一般建议，当您瞄准结构化并发时，我建议首先编写同步代码，然后再并发运行它们。

    • 0

相关问题