大话并发

严谨的并发应可静态推导出结果：根据线程内顺序一致性，结合Channel或sync同步事件的可排序性来推导，最终完成各个线程各段代码的顺序关系排序。如果两个事件无法根据规则来排序，显然是并发，执行先后顺序亦不可靠。

不靠谱的同步

新手初入golang，常有类似如下代码

func main() {
    go println("hello, world")
    time.Sleep(time.Second)
}

看似没问题，假设println函数内部实现休眠的时间大于main线程休眠的时间的话，就会导致矛盾：后台线程既然先于main线程完成打印，那么执行时间肯定是小于main线程执行时间的。当然这是不可能的。严谨的并发程序的正确性不应该是依赖于CPU的执行速度和休眠时间等不靠谱的因素。

互斥锁登场

在main函数所在线程中执行两次mu.Lock()，当第二次加锁时会因为锁已经被占用而阻塞，main函数的阻塞状态驱动后台线程继续向前执行。

func main() {
    var mu sync.Mutex

    mu.Lock()
    go func(){
        fmt.Println("你好, 世界")
        mu.Unlock()
    }()
    mu.Lock()
}

要等待n个线程完成后，再进行下一步同步操作，简易做法，通常是使用sync.WaitGroup来等待一组事件

func main() {
    var wg sync.WaitGroup
    // 开N个后台打印线程
    for i := 0; i < 10; i++ {
        wg.Add(1)

        go func() {
            fmt.Println("你好, 世界")
            wg.Done()
        }()
    }
    // 等待N个后台线程完成
    wg.Wait()
}

生产消费

带缓冲的通道在生产/消费模式下使用。通道好比一个中转仓，生产产品过多，仓库装不下，只得暂停生产（通道写入阻塞）。同样的道理，仓库没货，消费者也不可能有产品消费（通道读取阻塞）。在此二种情况下，二者相互依赖。

让main函数保存阻塞状态不退出，只有当用户输入Ctrl-C时才真正退出程序

// 生产者: 生成 factor 整数倍的序列
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) // 生成 3 的倍数的序列
    go Producer(5, ch) // 生成 5 的倍数的序列
    go Consumer(ch)    // 消费 生成的队列

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

发布订阅

发布订阅模型则是将消息发布给一个主题

package main

import (
    "sync"
    "time"
)

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

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

// 构建一个发布者对象, 可以设置发布超时时间和缓存队列的长度
func NewPublisher(publishTimeout time.Duration, buffer int) *Publisher {
    return &Publisher{
        buffer:      buffer,
        timeout:     publishTimeout,
        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()
    defer 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)
    }
    wg.Wait()
}

// 关闭发布者对象，同时关闭所有的订阅者管道。
func (p *Publisher) Close() {
    p.m.Lock()
    defer 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):
    }
}

下面展示了两个不同的订阅者，一个全部订阅与一个包含golang的订阅者

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

达者优先

用户不会关注你的算法，代码写的如何好。它只关注给出了指令，能否最快的时间得到它想的结果，比之如搜索。

示例创建了一个带缓存的通道，管道的缓存数目要足够大，保证不会因为缓存的容量引起不必要的阻塞。然后开启了多个后台线程，分别向不同的搜索引擎提交搜索请求。当任意一个搜索引擎最先有结果之后，都会马上将结果发到通道中（因为通道带了足够的缓存，这个过程不会阻塞）。但是最终只从通道取最先返回的结果，其它忽视。

func main() {
    ch := make(chan string, 32)

    go func() {
        ch <- searchByBing("golang")
    }()
    go func() {
        ch <- searchByGoogle("golang")
    }()
    go func() {
        ch <- searchByBaidu("golang")
    }()

    fmt.Println(<-ch)
}

通过适当开启一些冗余的线程，尝试用不同途径去解决同样的问题，最终以达者优先的方式提升了程序的相应性能

本作品采用《CC 协议》，转载必须注明作者和本文链接