我試圖了解如何正確使用 goroutines 以及通道和上下文，以創(chuàng)建可取消的后臺工作者。我熟悉使用在顯式調用時可以取消的上下文，將它附加到 worker goroutine 應該可以讓我停止 worker。但我無法弄清楚如何使用它來實現這一目標。下面的示例說明了一個從通道“urls”獲取數據的 worker goroutine，它還帶有一個可取消的上下文。//worker.gofunc Worker(id int, client *http.Client, urls chan string, ctx context.Context, wg *sync.WaitGroup) { fmt.Printf("Worker %d is starting\n", id) select { // placeholder for a channel writing the data from the URL case url := <-urls: fmt.Printf("Worker :%d received url :%s\n", id, url) // checking if the process is cancelled case <-ctx.Done(): fmt.Printf("Worker :%d exitting..\n", id) } fmt.Printf("Worker :%d done..\n", id) wg.Done()}這對我不起作用有兩個原因，對于無緩沖的通道，在沒有 goroutines 讀取的情況下寫入它會阻塞它，所以一旦有更多數據添加到 urls 通道，發(fā)送方就會阻塞。一旦兩個通道中的任何一個返回，它就會立即返回。我還嘗試將選擇包裝在一個無限循環(huán)中，但在上下文引發(fā)錯誤后添加一個中斷。func Worker(id int, client *http.Client, urls chan string, ctx context.Context, wg *sync.WaitGroup) { fmt.Printf("Worker %d is starting\n", id) for { select { // placeholder for a channel writing the data from the URL case url := <-urls: fmt.Printf("Worker :%d received url :%s\n", id, url) // checking if the process is cancelled case <-ctx.Done(): fmt.Printf("Worker :%d exitting..\n", id) break // raises error :ineffective break statement. Did you mean to break out of the outer loop? (SA4011)go-staticcheck } } fmt.Printf("Worker :%d done..\n", id) // code is unreachable wg.Done()}實施這樣的事情的正確方法是什么？PS：有關設計此類工作進程的任何資源/參考資料也將有所幫助。
2 回答
人到中年有點甜
TA貢獻1895條經驗 獲得超7個贊
您可以用 return 代替 break，代碼將起作用。
但是，更好的方法可能是：
Worker 在 for / range 循環(huán)中消費通道
生產者應負責檢測取消并關閉通道。for 循環(huán)將停止級聯
不負相思意
TA貢獻1777條經驗 獲得超10個贊
我專門為此做了一個 Go 包。你可以在這里找到它：https ://github.com/MicahParks/ctxerrpool
這是項目的示例README.md：
package main
import (
"bytes"
"context"
"log"
"net/http"
"os"
"time"
"github.com/MicahParks/ctxerrpool"
)
func main() {
// Create an error handler that logs all errors.
var errorHandler ctxerrpool.ErrorHandler
errorHandler = func(pool ctxerrpool.Pool, err error) {
log.Printf("An error occurred. Error: \"%s\".\n", err.Error())
}
// Create a worker pool with 4 workers.
pool := ctxerrpool.New(4, errorHandler)
// Create some variables to inherit through a closure.
httpClient := &http.Client{}
u := "https://golang.org"
logger := log.New(os.Stdout, "status codes: ", 0)
// Create the worker function.
var work ctxerrpool.Work
work = func(ctx context.Context) (err error) {
// Create the HTTP request.
var req *http.Request
if req, err = http.NewRequestWithContext(ctx, http.MethodGet, u, bytes.NewReader(nil)); err != nil {
return err
}
// Do the HTTP request.
var resp *http.Response
if resp, err = httpClient.Do(req); err != nil {
return err
}
// Log the status code.
logger.Println(resp.StatusCode)
return nil
}
// Do the work 16 times.
for i := 0; i < 16; i++ {
// Create a context for the work.
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
// Send the work to the pool.
pool.AddWorkItem(ctx, work)
}
// Wait for the pool to finish.
pool.Wait()
}
- 2 回答
- 0 關注
- 231 瀏覽
添加回答
舉報
0/150
提交
取消