mirror/go-patterns
1
0
Fork 0
mirror of https://github.com/tmrts/go-patterns.git synced 2024-11-29 00:25:31 +03:00
Code Issues Projects Releases Wiki Activity

Producer Consumer Part

Browse Source 
Like serverAccpet, create routine for consumer if there is available
consumer.
This commit is contained in:
Wei Fu 2017-03-30 21:21:58 +08:00
parent abba91a6ac
commit c1b89ad941
2 changed files with 16 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
concurrency/producer_consumer/common.go
View File

@ -15,7 +15,7 @@ const (
TimeoutDial = 5 * time.Second TimeoutDial = 5 * time.Second
TimeoutShutdown = 5 * time.Second TimeoutShutdown = 5 * time.Second
ProcessDuration = 2 * time.Second ProcessDuration = 3 * time.Second
) )
var ( var (

28
concurrency/producer_consumer/producer.go
View File

@ -54,19 +54,21 @@ func (p *Producer) schedule() {
// one consumer consumes one job at one time // one consumer consumes one job at one time
consumer := <-p.consumers consumer := <-p.consumers
// if rpcCall fails, this consumer will be regarded as unavailable. go func(consumer string) {
err := rpcCall(consumer, "Consumer.DoTask", task, &struct{}{}) // if rpcCall fails, this consumer will be regarded as unavailable.
if err != nil { err := rpcCall(consumer, "Consumer.DoTask", task, &struct{}{})
LogError.Printf("[%s] %s", p.address, err.Error()) if err != nil {
} else { // re-register consumer LogError.Printf("[%s] %s", p.address, err.Error())
go func(consumer string) { } else { // re-register consumer
select { go func(consumer string) {
case p.consumers <- consumer: select {
case <-time.After(5 * time.Second): case p.consumers <- consumer:
LogError.Printf("[%s] %s", p.address, ErrorTRegister) case <-time.After(5 * time.Second):
} LogError.Printf("[%s] %s", p.address, ErrorTRegister)
}(consumer) }
} }(consumer)
}
}(consumer)
case <-p.done: case <-p.done:
break break
} }