【Go】Go语言并发设计思路及来源

Go并发编程核心的CSP理论的核心概念只有一个：同步通信。 —— Go语言高级编程

写这篇文章的目的，就是想了解 Go语言并发的设计思路，想法，精髓，体验语言的美妙。

《Go语言圣经》中给出的Go语言的基因图谱，介绍了对Go语言产生影响的语言。

1. CSP并发模型

Go语言标志性的并发编程特性，来自于贝尔实验室1978年发表的关于并发研究的基础文献：顺序通信进程（communicating sequential processes，缩写为CSP）。

在最初的CSP论文中，程序只是一组没有中间共享状态的平行运行的处理过程，它们之间使用管道进行通信和控制同步。

CSP并发模型最经典的实际应用是爱立信发明的Erlang编程语言。

2. Newsqueak语言

这门语言是Rob Pick发明的squeak语言的第二代，是他用于实践CSP并发编程模型的战场。

Squeak语言是一个提供鼠标和键盘事件处理的编程语言，它的管道(channel)是静态创建的。

改进版的Newsqueak语言提供了类似C语言语句的表达式语法、类似Pascal语言的推导语法，它是一个带自动垃圾回收的纯函数语言。

在Newsqueak中，管道是动态创建的，可以保存在变量中。

我们通过“素数筛法”来理解Newsqueak语法对Go语言产生的影响，其原理如图：

通过一个又一个的管道传输数据、过滤数据，最后得到的就是一个又一个的素数。

3. Alef语言

Alef语言对线程和进程的并发体都提供了支持，它们通过管道进行通讯。

不过由于Alef缺乏内存自动回收机制，导致并发体的内存资源管理异常复杂。

可以将Alef理解成增强并发的C语言。

4. Limbo语言

它是运行在小型计算机上的分布式应用的编程语言，它支持模块化编程，编译期和运行时的强类型检查。

它其实已经具备了Go语言的雏形。

可以看到，Go语言的并发，其实也是在前人的基础上一步一步优化过来的。

2、生产者消费者模型

并发编程中最常见的例子就是生产者消费者模式，该模式主要通过平衡生产线程和消费线程的工作能力来提高整体处理数据的速度。

package main

import (
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"
)

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

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

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

	go Producer(3, ch)
	go Producer(5, ch)
	go Consumer(ch)

	time.Sleep(5 * time.Second)

	// Ctrl+C 退出程序
	sig := make(chan os.Signal, 1)
	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
	fmt.Printf("quit (%v)\n", <-sig)
}

package main

import (

"fmt"

"os"

"os/signal"

"syscall"

"time"

)

func Producer(factor int, out chan<- int) {

for i := 0; ; i++ {

out <- i * factor

}

func Consumer(in <-chan int) {

for v := range in {

fmt.Println(v)

}

func main() {

ch := make(chan int, 64)

go Producer(3, ch)

go Producer(5, ch)

go Consumer(ch)

time.Sleep(5 * time.Second)

// Ctrl+C 退出程序

sig := make(chan os.Signal, 1)

signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)

fmt.Printf("quit (%v)\n", <-sig)

}

上面是一个生产者消费者模型，有两个生产者，分别生产3和5的倍数，消费者打印输出。

3、发布订阅模型

发布订阅模型（publish-and-subscribe）模型，通常被简写为pub/sub模型。

在这个模型中，消息的生产者是发布者，消息消费者是订阅者，生产者和消费者是M:N的关系。

//Package pubsub implements a simple multi-topic pub-sub library.
package pubsub

import (
	"sync"
	"time"
)

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

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

//创建一个发布者对象, 可以设置发布超时时间和缓存队列的长度
func NewPublisher(publishTimeout time.Duration, buffer int) *Publisher {
	return &Publisher{
		buffer:     buffer,
		timeout:    publishTimeout,
		subscriber: 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.subscriber[ch] = topic
	p.m.Unlock()
	return ch
}

//退出订阅
func (p *Publisher) Evict(sub chan interface{}) {
	p.m.Lock()
	defer p.m.Unlock()
	delete(p.subscriber, 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.subscriber {
		wg.Add(1)
		go p.sendTopic(sub, topic, v, &wg)
	}
	wg.Wait()
}

//关闭发布者对象，同时关闭所有的订阅者通道。
func (p *Publisher) Close() {
	p.m.Lock()
	defer p.m.Unlock()

	for sub := range p.subscriber {
		delete(p.subscriber, 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):
	}
}

//Package pubsub implements a simple multi-topic pub-sub library.

package pubsub

import (

"sync"

"time"

)

type (

subscriber chan interface{} // 订阅者为一个管道

topicFunc func(v interface{}) bool // 主题为一个过滤器

)

type Publisher struct {

m sync.RWMutex //读写锁

buffer int //订阅队列的缓存大小

timeout time.Duration //发布超时时间

subscriber map[subscriber]topicFunc //订阅者信息

}

//创建一个发布者对象, 可以设置发布超时时间和缓存队列的长度

func NewPublisher(publishTimeout time.Duration, buffer int) *Publisher {

return &Publisher{

buffer: buffer,

timeout: publishTimeout,

subscriber: 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.subscriber[ch] = topic

p.m.Unlock()

return ch

}

//退出订阅

func (p *Publisher) Evict(sub chan interface{}) {

p.m.Lock()

defer p.m.Unlock()

delete(p.subscriber, 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.subscriber {

wg.Add(1)

go p.sendTopic(sub, topic, v, &wg)

}

wg.Wait()

}

//关闭发布者对象，同时关闭所有的订阅者通道。

func (p *Publisher) Close() {

p.m.Lock()

defer p.m.Unlock()

for sub := range p.subscriber {

delete(p.subscriber, 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):

}

接着就是主程序：

package main

import (
    "AdvancedGo/ch1/pubsub"
    "fmt"
    "strings"
    "time"
)

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
    })

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

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

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

    time.Sleep(3 * time.Second)
}

package main

import (

"AdvancedGo/ch1/pubsub"

"fmt"

"strings"

"time"

)

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

})

p.Publish("hello, world!")

p.Publish("hello, golang!")

go func() {

for msg := range all {

fmt.Println("all:", msg)

}

}()

go func() {

for msg := range golang {

fmt.Println("golang:", msg)

}

}()

time.Sleep(3 * time.Second)

}

【Go】Go语言并发设计思路及来源

Tagged on: Go