Goroutine Pool

 1package test
 2
 3import (
 4	"fmt"
 5	"runtime"
 6	"testing"
 7	"time"
 8)
 9
10type PPool struct {
11	EntryChan chan func() error
12	JobChan   chan func() error
13	WorkNum   int
14}
15
16func NewPPool(cap int) *PPool {
17	p := PPool{
18		EntryChan: make(chan func() error),
19		JobChan:   make(chan func() error),
20		WorkNum:   cap,
21	}
22	return &p
23}
24
25func (p *PPool) worker(wId int) {
26	for t := range p.JobChan {
27		err := t()
28		fmt.Println("workerId", wId, "执行任务成功")
29		if err != nil {
30			fmt.Println(err)
31		}
32	}
33}
34
35func (p *PPool) Receive(t func() error) {
36	p.EntryChan <- t
37}
38
39func (p *PPool) Run() {
40	for i := 0; i < p.WorkNum; i++ {
41		go p.worker(i)
42	}
43	for t := range p.EntryChan {
44		p.JobChan <- t
45	}
46	close(p.JobChan)
47}
48
49func TestPool(t *testing.T) {
50	tt := func() error {
51		fmt.Println(time.Now())
52		return nil
53	}
54
55	p := NewPPool(3)
56	go func() {
57		i := 1
58		for {
59			p.Receive(tt)
60			i++
61		}
62		// close by send side
63		close(p.EntryChan)
64	}()
65	p.Run()
66}
67
68func loop(num int, quit chan int) {
69	for i := 0; i < 10; i++ {
70		runtime.Gosched()
71		fmt.Printf("%d-%d\n", num, i)
72	}
73	quit <- 0
74}
75
76func TestGo(t *testing.T) {
77	fmt.Println(runtime.NumCPU())
78	var quit chan int = make(chan int)
79	go loop(1, quit)
80	go loop(2, quit)
81	go loop(3, quit)
82
83	for i := 0; i < 3; i++ {
84		<-quit
85	}
86}