我是新手，找不到這個問題的答案。我正在做的是在生產者中讀取 CSV 文件，做一些可能需要時間的事情，然后通過通道將輸出發送給消費者。有一個生產者-消費者鏈，任何生產者最終都可能比它的消費者慢。producer(1 goroutine) -> chan0 -> consumer-producer-1(>1 goroutines) -> chan1 -> consumer-producer-2(>1 goroutines) -> chan2 -> consumer(>1 goroutines)這里最多可以有 15 個消費者?，F在我面臨的問題是，如果生產者完成了，如何在消費者方面做出決定，我們可以停止處理。我需要實現的是：一旦生產者完成，所有消費者最終應該做一些清理并在完成剩余的后退出如果消費者在特定的超時期限內沒有獲得任何數據，它可以退出（最好有一個信號）而不會進一步阻塞。它發生在序列中的所有生產者-消費者對上。我使用了以下方法。將信號通道與每個數據通道保持在一起，并為其下一個消費者的每個 goroutine 發布“完成”。讀取后，每個消費者應該只讀取通道中剩余的緩沖數據，然后將 5“完成”放在下一個信號通道上。確保每個 goroutine 只有 5 個，而不是 5 個（使用https://golang.org/pkg/sync/#Once.Do）。以下是我到這里為止能想到的。processRemaining = falsefor processRemaining == false{        select {        case stuff, ok := <-input_messages:                do_stuff(stuff)                if ok == false { // if channel has been closed                    processRemaining = true                }                if result != nil {                        //send to channel output_messages                }        case sig := <-input_signals: // if signaled to stopped.                fmt.Println("received signal", sig)                processRemaining = true        default:                fmt.Println("no activity")        }}if processRemaining {        for stuff := range input_messages {                do_stuff(stuff)                if result != nil {                        //send to channel output_messages                }        }        // send "output_routine" number of "done" to a channel "output_signals".}但即使在這種方法中，我也無法想出任何方式來表現與關閉的“input_messages”通道相同的方式，如果沒有可用的時間，比如 10 秒。這種方法有沒有我忽略的問題。解決這個問題的可能方法（或并發模式）是什么？確保：一旦第一個“chan0”關閉，所有后續通道都會關閉。所有的生產者在關閉他們的輸出通道之前都會被更新，只有當他們都完成了他們的寫入時，通道才會關閉。如果消費者在指定的超時時間內沒有從通道中獲取數據，它應該將其視為已關閉，并自行解除阻塞。