Golang Nicht Blocking Buffer

Question

Synchron Beispiel:

type job struct { 
    Id int 
    Message string 
} 

for { 
    // getJob() blocks until job is received 
    job := getJob() 
    doSomethingWithJob(job) 
} 

Ich wünsche Jobs zu verarbeiten, wie sie in von getJob mit doSomethingWithJob kommen. z.B. getJob könnte eine Payload sein, die von einer MessagingQueue wie RabbitMQ/Beanstalkd oder einer HTTP-Anfrage empfangen wird.

Ich möchte getJob nicht blockieren, während ich bin doSomethingWithJob & umgekehrt. Ich möchte jedoch die Anzahl der Jobs kontrollieren/puffern, damit ich das System nicht überlasten kann. z.B. max Gleichzeitigkeit von 5.

Das Konzept der Go-Routinen verwirren mich im Moment, so dass alle Zeiger in die richtige Richtung würde geschätzt viel zu helfen mir zu lernen.

Aktualisieren: Danke @JimB für Ihre Hilfe. Warum nimmt Arbeiter 5 immer den Job auf?

jobCh := make(chan *job) 

// Max 5 Workers 
for i := 0; i < 5; i++ { 

    go func() { 

     for job := range jobCh { 
      time.Sleep(time.Second * time.Duration(rand.Intn(3))) 
      log.Println(i, string(job.Message)) 
     } 
    }() 
} 

for { 
    job, err := getJob() 
    if err != nil { 
     log.Println("Closing Channel") 
     close(jobCh) 
     break 
    } 

    jobCh <- job 
} 

log.Println("Complete") 

Beispiel Ausgabe

2016/06/09 22:19:57 5 {"id":10692,"name":"Test Message"} 
2016/06/09 22:19:57 5 {"id":10687,"name":"Test Message"} 
2016/06/09 22:19:57 5 {"id":10699,"name":"Test Message"} 
2016/06/09 22:19:57 5 {"id":10701,"name":"Test Message"} 
2016/06/09 22:19:57 5 {"id":10703,"name":"Test Message"} 
2016/06/09 22:19:57 5 {"id":10704,"name":"Test Message"} 

Answer 1

Sie können 5 goroutines von einem Kanal zu lesen beginnen doSomethingWithJob zu nennen. Auf diese Weise werden nie mehr als 5 Jobs gleichzeitig verarbeitet.

jobCh := make(chan *job) 

// start 5 workers to process jobs 
for i := 0; i < 5; i++ { 
    go func() { 
     for job := range jobCh { 
      doSomethingWithJob(job) 
     } 
    }() 
} 

// send jobs to workers as fast as we can 
for { 
    jobCh <- getJob() 
}