goroutine pool

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

type worker struct {
    Func func()
}

func main() {
    var wg sync.WaitGroup

    channels := make(chan worker, 10)

    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for ch := range channels {
                //reflect.ValueOf(ch.Func).Call(ch.Args)
                ch.Func()
            }
        }()
    }

    for i := 0; i < 100; i++ {
        j := i
        wk := worker{
            Func: func() {
                fmt.Println(j + j)
            },
        }
        channels <- wk
    }
    close(channels)
    wg.Wait()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

package pool

type PayloadProcessor func(payload interface{})

type Pool struct {
    Size      int
    Channels  chan interface{}
    Processor PayloadProcessor
}

func New(size int, processor func(interface{})) *Pool {
    p := Pool{
        Size:      size,
        Channels:  make(chan interface{}, 0),
        Processor: processor,
    }

    for i := 0; i < p.Size; i++ {
        go func() {
            for payload := range p.Channels {
                p.Processor(payload)
            }
        }()
    }
    return &p
}

func (p *Pool) Process(payload interface{}) {
    p.Channels <- payload
}