golang的并发模型

 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

func main() {
	ch := make(chan int, 64)

	go Producer(2, ch)
	go Producer(3, ch)

	go Consumer(ch)

	sig := make(chan os.Signal, 1)
	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
	fmt.Printf("quit(%v)\n", <-sig)
}

//Producer 生产者
func Producer(factor int, out chan<- int) {
	for i := 0; ; i++ {
		out <- i * factor
	}
}

//Consumer 消费者
func Consumer(in <-chan int) {
	for v := range in {
		fmt.Println(v)
	}
}

 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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

package pubsub

import (
	"sync"
	"time"
)

type (
	subscriber chan interface{}         //订阅者位一个通道
	topicFunc  func(v interface{}) bool //主题为一个过滤器
)

//Publisher 发布者对象
type Publisher struct {
	m           sync.RWMutex             //读写锁
	buffer      int                      //订阅队列的缓存大小
	timeout     time.Duration            //发布的超时时间
	subscribers map[subscriber]topicFunc //订阅者信息
}

//构建一个发布者对象，可以设置发布超时时间和缓存队列的长度
func NewPublisher(publisherTimeOut time.Duration, buffer int) *Publisher {
	return &Publisher{
		buffer:      buffer,
		timeout:     publisherTimeOut,
		subscribers: make(map[subscriber]topicFunc),
	}
}

//订阅全部
func (p *Publisher) Subscribe() chan interface{} {
	return p.SubscribeTopic(nil)
}

//添加订阅
func (p *Publisher) SubscribeTopic(topic topicFunc) chan interface{} {
	ch := make(chan interface{}, p.buffer)
	p.m.Lock()
	p.subscribers[ch] = topic
	p.m.Unlock()
	return ch
}

//退订
func (p *Publisher) Evict(sub chan interface{}) {
	p.m.Lock()
	p.m.Unlock()

	delete(p.subscribers, sub)
	close(sub)
}

//发布主题
func (p *Publisher) Publish(v interface{}) {
	p.m.RLock()
	defer p.m.RUnlock()

	var wg sync.WaitGroup
	for sub, topic := range p.subscribers {
		wg.Add(1)
		go p.sendTopic(sub, topic, v, &wg)
	}

}

//关闭发布者对象,关闭所有订阅者通道
func (p *Publisher) Close() {
	p.m.Lock()
	p.m.Unlock()

	for sub := range p.subscribers {
		delete(p.subscribers, sub)
		close(sub)
	}
}

//发送主题
func (p *Publisher) sendTopic(sub subscriber, topic topicFunc, v interface{}, wg *sync.WaitGroup) {
	defer wg.Done()
	if topic != nil && !topic(v) {
		return
	}
	select {
	case sub <- v:
	case <-time.After(p.timeout):
	}

}

 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
33
34
35
36
37
38
39
40
41
42
43
44
45
46

func main() {
	p := pubsub.NewPublisher(100*time.Millisecond, 10)
	defer p.Close()

	all := p.Subscribe()

	golang := p.SubscribeTopic(func(v interface{}) bool {
		if s, ok := v.(string); ok {
			return strings.Contains(s, "golang")
		}
		return false
	})

	php := p.SubscribeTopic(func(v interface{}) bool {
		if s, ok := v.(string); ok {
			return strings.Contains(s, "php")
		}
		return false
	})

	p.Publish("hello,  world!")
	p.Publish("hello, golang!")
	p.Publish("hello, php!")

	go func() {
		for msg := range all {
			fmt.Println("all:", msg)
		}
	}()

	go func() {
		for msg := range golang {
			fmt.Println("golang:", msg)
		}
	}()

	go func() {
		for msg := range php {
			fmt.Println("php:", msg)
		}
	}()

	// 运行一定时间后退出
	time.Sleep(3 * time.Second)
}