Goroutinen durch For-Schleife blockiert?

Question

Ich habe den folgenden Code, der einen Arbeitswarteschlange implementiert:

package main 

import (
    "fmt" 
    "net/http" 
    "io" 
    "time" 
) 

var (
    linkQueue chan Link 
    scraperQueue chan chan Link 
) 

func CycleDirectory(page int) { 
    linkQueue <- Link{Name: "asd"} 
} 

type Link struct { 
    Name string 
} 

func (s Scraper) Start() { 
    fmt.Println("Started") 
    go func() { 
     for { 
      s.ScraperQueue <- s.Link 
      select { 
      case link := <-s.Link: 
       fmt.Println(fmt.Sprintf("%v", s.Id) + ": Received " + link.Name) 
      case <-s.QuitChan: 
       fmt.Println("Closed") 
       return 
      } 
     } 
    }() 
} 

func (s Scraper) Stop() { 
    go func() { 
     s.QuitChan <- true 
    }() 
} 

type Scraper struct { 
    Id int 
    Link chan Link 
    ScraperQueue chan chan Link 
    QuitChan chan bool 
} 

func InitScraper(id int, scraperQueue chan chan Link) Scraper { 
    return Scraper { 
     Id: id, 
     Link: make(chan Link), 
     ScraperQueue: scraperQueue, 
     QuitChan: make(chan bool), 
    } 
} 

func HelloServer(w http.ResponseWriter, req *http.Request) { 
    io.WriteString(w, "hello, world!\n") 
} 

func main() { 
    linkQueue = make(chan Link, 2000) 

    numScrapers := 2 

    scraperQueue = make(chan chan Link, numScrapers) 

    for i := 0; i < numScrapers; i++ { 
     s := InitScraper(i+1, scraperQueue) 
     s.Start() 
    } 

    go func() { 
     for { 
      select { 
      case link := <-linkQueue: 
       go func() { 
        scraper := <-scraperQueue 
        scraper <- link 
       }() 
      } 
     } 
    }() 

    CycleDirectory(1) 

    // time.Sleep(1 * time.Millisecond) 

    for { 
     // select { 
     // } 
    } 

    // http.HandleFunc("/hello", HelloServer) 

    // http.ListenAndServe(":12345", nil) 
} 

Ausführen dieses Codes einer for-Schleife enthält eine if-Anweisung (oder nichts innen), ist der Abstreifer nicht eine empfangene Nachricht drucken. Mit der ListenAndServe-Funktion von net/http wird die empfangene Nachricht gedruckt. Blockiert mit Schlaf für 1 ms, erhalte ich die Nachricht. Und indem ich eine select-Anweisung in die for-Schleife setze, erhalte ich auch die Nachricht.

Warum ist die for-Schleife ohne eine Select-Anweisung nicht für die Ausführung der Nachricht senden in den Worker-Warteschlangen, und wie würde ich gehen, dies zu behandeln. Ich brauche eine if-Anweisung in der for-Schleife, um zu überprüfen, ob die ganze Arbeit erledigt wurde, damit ich die Schleife verlassen und das Programm beenden kann.

aktualisieren

AMDs Vorschlag ist, eine Lösung für dieses Problem. Hier ist meine aktualisierte Code sync.WaitGroup Paket Haupt

import (
    "fmt" 
    "sync" 
) 

var (
    linkQueue chan Link 
    scraperQueue chan chan Link 
    wg sync.WaitGroup 
) 

func CycleDirectory(page int) { 
    wg.Add(1) 
    linkQueue <- Link{Name: "asd"} 
} 

type Link struct { 
    Name string 
} 

func (s Scraper) Start() { 
    fmt.Println("Started") 
    go func() { 
     for { 
      s.ScraperQueue <- s.Link 
      select { 
      case link := <-s.Link: 
       Scrape(s.Id, link.Name) 
       s.Stop() 
      case <-s.QuitChan: 
       fmt.Println("Closed") 
       wg.Done() 
       return 
      } 
     } 
    }() 
} 

func (s Scraper) Stop() { 
    go func() { 
     s.QuitChan <- true 
    }() 
} 

type Scraper struct { 
    Id int 
    Link chan Link 
    ScraperQueue chan chan Link 
    QuitChan chan bool 
} 

func Scrape(id int, name string) { 
    fmt.Println(fmt.Sprintf("%v", id) + ": Received " + name) 
} 

func InitScraper(id int, scraperQueue chan chan Link) Scraper { 
    return Scraper { 
     Id: id, 
     Link: make(chan Link), 
     ScraperQueue: scraperQueue, 
     QuitChan: make(chan bool), 
    } 
} 

func main() { 
    linkQueue = make(chan Link, 2000) 

    numScrapers := 2 

    scraperQueue = make(chan chan Link, numScrapers) 

    for i := 0; i < numScrapers; i++ { 
     s := InitScraper(i+1, scraperQueue) 
     s.Start() 
    } 

    go func() { 
     for { 
      select { 
      case link := <-linkQueue: 
       go func() { 
        scraper := <-scraperQueue 
        scraper <- link 
       }() 
      } 
     } 
    }() 

    CycleDirectory(1) 

    wg.Wait() 

    fmt.Println("Done") 
} 

Answer 1

verwenden, können Sie sync.WaitGroup verwenden um das Programm zu stoppen Verlassen, bis alle die Arbeit getan ist.
Versuchen Sie es auf The Go Playground:

package main 

import (
    "fmt" 
    "sync" 
    "time" 
) 

var (
    wg sync.WaitGroup 
) 

func main() { 
    wg.Add(1) 
    go func() { 
     defer wg.Done() 
     time.Sleep(2 * time.Second) 
    }() 

    fmt.Println("Wait...") 
    wg.Wait() 
    fmt.Println("Done.") 
}