主页 / coding / 问题 / 79151872
Accepted
Ahmed Zaidan
Asked: 2024-11-03 09:05:46 +0800 CST

Go 退出嵌入的 goroutines

  • 772

在 Go 中我无法弄清楚如何获得以下设置。

  • 第一级 goroutine 应该在终止之前停止其第二级 goroutine。
  • 如果函数runGoroutine没有循环（我不希望它循环），那么当消息发送到通道时，goroutine 将如何退出？如果没有循环，那么case <-stopChan:只会在开始时被调用一次。
  • 如果函数runGoroutine调用其他异步函数会怎么样？当返回时这些函数也会退出吗？

也许我的方法不对。在我的情况下，我有x用户可以启动的任务组，当用户启动任务组时，会创建一个 goroutine（第 1 级）以返回 CL 用户访问权限。此任务组（第 1 级）是yAPI 请求的集合（实际上 y 是 200-300）。对于每个 API 请求，我都会创建一个新的 go-routine（第 2 级）。我的目标是速度；如果我每秒执行数百个 API 请求，将每个请求放入其自己的 goroutine 中是否更快/更有效？或者在 1 个线程上执行是否相同？

                       Parent/Main process
                        |              |
          Child goroutine            Child goroutine
                |                           |
      X bottom goroutines          X bottom goroutines           (API level)
 

package main

import (
    "fmt"
    "time"
)

func runGoroutine(stopChan <-chan struct{}) {
    for {
        select {
        case <-stopChan:
            return // Exit the goroutine
        default:
            fmt.Println("Goroutine is working...")
            time.Sleep(1 * time.Second)
        }
    }
}
func main() {
    stopChan := make(chan struct{})
    // Start the goroutine
    go runGoroutine(stopChan)
    // User should be able to close the routines they want
    close(stopChan)
}

新代码

package main

import (
    "context"
    "log"
    "net/http"
    "sync"
    "time"
)

func main() {
    // Initialize context with cancellation
    ctx, cancel := context.WithCancel(context.Background())

    // Call the worker pool in a separate goroutine to allow main to return immediately

    // We can start any task group like this

    go startWorkerPool(ctx, 20, 100)
    
    // User has CL access on main thread
    // Main function returns, but `cancel` is accessible to allow later cancellation if needed
    log.Println("Worker pool started; you can cancel it by calling cancel()")
    waitForNine()
}

func waitForNine() {
    var input int
    for input != 9 {
        fmt.Print("Enter a number: ")
        fmt.Scan(&input)
    }
}

// startWorkerPool starts a fixed-size worker pool to handle `numRequests` requests.
func startWorkerPool(ctx context.Context, numWorkers, numRequests int) {
    var wg sync.WaitGroup
    work := make(chan string)

    // Launch the specified number of workers
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(ctx context.Context) {
            defer wg.Done()
            client := http.Client{}
            for url := range work {
                if req, err := http.NewRequestWithContext(ctx, "GET", url, nil); err != nil {
                    log.Println("failed to create new request:", err)
                } else if resp, err := client.Do(req); err != nil {
                    log.Println("failed to make request:", err)
                } else {
                    resp.Body.Close()
                }
            }
        }(ctx)
    }

    // Enqueue URLs for workers and close the channel when done
    go func() {
        for i := 0; i < numRequests; i++ {
            work <- "https://httpbin.org/get"
        }
        close(work)
    }()

    // Wait for all workers to finish
    wg.Wait()
    log.Println("All requests processed")
}
multithreading

1 个回答

  1. Best Answer

    共享context是一次性取消多个 goroutine 的简单方法。考虑这个简单的例子：

    package main

import (
    "context"
    "log"
    "sync"
    "time"
    "net/http"
)

func main() {
    var wg sync.WaitGroup
    ctx, cancel := context.WithCancel(context.Background())
    go func() {
        time.Sleep(1 * time.Second)
        log.Print("Cancelling")
        cancel()
    }()
    for range 100 {
        wg.Add(1)
        go func(ctx context.Context) {
            defer wg.Done()
            if req, err := http.NewRequestWithContext(ctx, "GET", "https://httpbin.org/get", nil); err != nil {
                log.Println("failed to create new request: ", err.Error())
            } else if resp, err := new(http.Client).Do(req); err != nil {
                log.Println("failed to make request: ", err.Error())
            } else {
                resp.Body.Close()
            }
        }(ctx)
    }
    log.Print("Waiting")
    wg.Wait()
}

    此代码启动 100 个并发 goroutine 来请求https://httpbin.org/get一个方便的测试 URL。程序启动一秒后，它们的共享上下文被取消，导致 http 请求中断。您可以使用信号或其他事件来取消共享上下文。

    如果我每秒执行数百个 API 请求，将每个请求放入其自己的 goroutine 中是否更快/更高效？

    启动 goroutine确实比通过通道传递值有更多开销。启动 http goroutine 的工作池并将工作分配给它们很有意义。

    下一个示例创建了 20 个 goroutine，这 20 个 goroutine 处理所有 100 个请求。当工作负载非常高时，工作池模式更为实用。

    package main

import (
    "context"
    "log"
    "net/http"
    "sync"
    "time"
)

func main() {
    var wg sync.WaitGroup
    ctx, cancel := context.WithCancel(context.Background())
    go func() {
        time.Sleep(1 * time.Second)
        log.Print("Cancelling")
        cancel()
    }()
    var work = make(chan string)
    for range 20 {
        wg.Add(1)
        go func(ctx context.Context) {
            defer wg.Done()
            client := http.Client{}
            for url := range work {
                if req, err := http.NewRequestWithContext(ctx, "GET", url, nil); err != nil {
                    log.Println("failed to create new request: ", err.Error())
                } else if resp, err := client.Do(req); err != nil {
                    log.Println("failed to make request: ", err.Error())
                } else {
                    resp.Body.Close()
                }
            }
        }(ctx)
    }
    go func() {
        for range 100 {
            work <- "https://httpbin.org/get"
        }
        close(work)
    }()
    log.Print("Waiting")
    wg.Wait()
}

    更新

    现在我们已经有了工作池，让我们让它从标准输入中拉取我们的 URL 列表。用户可以通过键入 URL 来提供 URL，或者通过将文件内容或其他来源重定向到标准输入来提供 URL。

    我们要做的就是开始从标准输入读取我们的 URL，并将它们传递到我们的工作队列以便在后台检索。

    我们一次只能处理这么多工作。过了这个点，我们可以选择让主机超负荷运转，或者暂停处理新工作；一般来说，后者更可取。在这个实现中，我们通过为 提供有限的缓冲区大小来限制我们排队的工作量work。

    package main

import (
    "bufio"
    "context"
    "log"
    "net/http"
    "net/url"
    "sync"
    "os"
)

func main() {
    var wg sync.WaitGroup
    ctx, cancel := context.WithCancel(context.Background())
    var work = make(chan string, 1000)
    for range 20 {
        wg.Add(1)
        go func(ctx context.Context) {
            defer wg.Done()
            client := http.Client{}
            for url := range work {
                if req, err := http.NewRequestWithContext(ctx, "GET", url, nil); err != nil {
                    log.Println("failed to create new request: ", err.Error())
                } else if resp, err := client.Do(req); err != nil {
                    log.Println("failed to make request: ", err.Error())
                } else {
                    resp.Body.Close()
                }
            }
        }(ctx)
    }
    wg.Add(1)
    go func() {
        defer wg.Done()
        //read standard input
        var scanner = bufio.NewScanner(os.Stdin)
        for scanner.Scan() {
            if scanner.Text() == "cancel" {
                // user could also just press Ctrl-C to kill this program and all its goroutines
                cancel()
                break
            }
            if url, err := url.Parse(scanner.Text()); err != nil {
                log.Printf("Not a valid url: %s", scanner.Text())
            } else {
                work <- url.String()
            }
        }
        if err := scanner.Err(); err != nil {
            log.Println("failed reading standard input: ", err.Error())
        }
        close(work)
    }()
    log.Print("Waiting")
    wg.Wait()
}

    您可以以交互方式运行程序，也可以从文件或其他来源提供输入。以下是在命令行提供 3 个 URL 并立即取消的示例。

    % echo -e "https://httpbin.org/get\nhttps://httpbin.org/get\nhttps://httpbin.org/get\ncancel" | ./tgo
2024/11/02 22:50:31 Waiting
2024/11/02 22:50:31 failed to make request:  Get "https://httpbin.org/get": context canceled
2024/11/02 22:50:31 failed to make request:  Get "https://httpbin.org/get": context canceled
2024/11/02 22:50:31 failed to make request:  Get "https://httpbin.org/get": context canceled

    如果省略cancel，程序仍会在处理完 URL 列表后结束，然后程序完成。这是因为scanner到达了文件末尾。您可以在命令行中使用 Ctrl-D 发出文件末尾信号。

    % ./tgo <<< "https://httpbin.org"
2024/11/02 22:53:33 Waiting
    • 1

相关问题