资料

并发和并行

并发 (Concurrency)：Composition of independently executing process (processes in the general sense, not linux processes.
并发指的是结构设计，并行是运行状态，有很多分解过程的思路，不同的分解思路，得到不同的并发组合设计。
并行 (Parallism): Simultaneous execution of computations

通过内存共享进行通讯与通过通讯来共享内存。

传统多线程模型：共享内存，锁，缺点：复杂，不可预测
CSP：传递数据和所有权，自动同步，不用面对共享内存和锁带来的复杂问题。

同步

初始化

If a package p imports package q, the completion of q's init functions happens before the start of any of p's.
The start of the function main.main happens after all init functions have finished.

Goroutine creation

go 语句用来起一个 goroutine，that starts a new goroutine happens before the goroutine's execution begins.
```
var a string

func f() {
    print(a)
}

func hello() {
    a = "hello, world"
    go f()
}
```

goroutine destruction

var a string

func hello() {
    go func() { a = "hello" }()
    print(a)
}

a 的赋值操作没有任何同步事件，所以其值不一定被其他 goroutine 看到。
Channel communication
A send on a channel happens before the corresponding receive from that channel completes. 这里没有区分 channel 的类型。
The closing of a channel happens before a receive that returns a zero value because the channel is closed.
A receive from an unbuffered channel happens before the send on that channel completes。这里区分了 channel 的类型，也就是说 send gorotine 会一直等待接受的 gorotine 能够接收
```
var c = make(chan int, 10)
var a string

func f() {
    a = "hello, world"
    c <- 0
}

func main() {
    go f()
    <-c
    print(a)  # 一定会 hello world
}
```

下面代码，如果 chanenl 是 buffered, c = make(chan int, 1)，那么不保证 a 打印"hello, world"。

var c = make(chan int)
var a string

func f() {
    a = "hello, world"
    <-c
}
func main() {
    go f()
    c <- 0
    print(a) # c 不是 buffer 的，那么利用 happens-before 的第二条规则，一定会输出 hello world
}

The kth receive on a channel with capacity C happens before the k+Cth send from that channel completes.

限制同时最多有 3 个 goroutine 同时执行

var limit = make(chan int, 3)

func main() {
    for _, w := range work {
        go func(w func()) {
            limit <- 1
            w()
            <-limit
        }(w)
    }
    select{}
}

Locks

For any sync.Mutex or sync.RWMutex variable l and n < m, call n of l.Unlock() happens before call m of l.Lock() returns.
For any call to l.RLock on a sync.RWMutex variable l, there is an n such that the l.RLock happens (returns) after call n to l.Unlock and the matching l.RUnlock happens before call n+1 to l.Lock.

Once

只初始化一次的语义：Multiple threads can execute once.Do(f) for a particular f, but only one will run f(), and the other calls block until f() has returned.

A single call of f() from once.Do(f) happens (returns) before any call of once.Do(f) returns.

var a string
var once sync.Once

func setup() {
    a = "hello, world"
}

func doprint() {
    once.Do(setup)
    print(a)
}

func twoprint() {
    go doprint()
    go doprint()
}

//只打印一次 hello world

不正确的同步

var a string
var done bool

func setup() {
    a = "hello, world"
    done = true
}

func main() {
    go setup()
    for !done {
    }
    print(a)
}

如何利用 select 和 channel 来表达优先级

for {
    select {
        case v := <-messageCh:
            fmt.Println(v)
        case <-disconnectCh:
            fmt.Println("disconnection, return")
        return
    }
}

for i := 0; i < 10; i++ {
    messageCh <- i
}
disconnectCh <- struct{}{}

// 执行的时候，可能出现如下的结果
0
1
2
3
4
disconnection, return

select 在处理多个 channel 都能够读的时候，会随机的选择一个，这是为了避免出现饥饿现象。
如何表达优先级呢？
- 将 messageCh 改成 unbuffer 的，那时如果在生产端有多个，那么会出现 data race
- 使用一个 channel，传递一个结构体，这个结构体中包含了一个 disconnectCh 的值。
- 如果确定要使用两个 channel，而且 messageCh 必须是有缓冲的，那么该如何处理呢？
- 能使用这样的语义来实现

实现优先级

for {
    select {
        case v := <-messageCh:
            fmt.Println(v)
        case <-disconnectCh:
            for {
                select {
                    case v := <-messageCh:
                        fmt.Println(v)
                    default:
                        fmt.Println("disconnection, return")
                    return
                }
            }
        }
    }
}

Receive from either messageCh or disconnectCh.
If a disconnection is received
- Read all the existing messages in messageCh, if any.
- Then return.
这里的 messageCh 读取不会阻塞等待，那么 default 不会马上执行

如何合并两个 read channel，返回一个 channel

func merge(ch1, ch2 <-chan int) <-chan int {
    ch := make(chan int, 1)
    go func() {
    for {
        select {
            case v := <-ch1:
                ch <- v
            case v := <-ch2:
                ch <- v
        }
    }
    close(ch)
}()
    return ch
}

这里的问题是如果 close 了 ch1，for 中的循环也是一直运行，因为 read a closed channel alway not block。

可能你想这样做

func merge(ch1, ch2 <-chan int) <-chan int {
    ch := make(chan int, 1)
    ch1Closed := false
    ch2Closed := false
    go func() {
        for {
            select {
            case v, open := <-ch1:
                if !open {
                    ch1Closed = true
                    break
                }
                ch <- v
            case v, open := <-ch2:
                if !open {
                    ch2Closed = true
                    break
                }
                ch <- v
            }
            if ch1Closed && ch2Closed {
                close(ch)
            return
        }
    }
}()
    return ch
}

一样的，有上面的问题。但是，读一个 nil 的 channel 将会 block.

func merge(ch1, ch2 <-chan int) <-chan int {
    ch := make(chan int, 1)
    go func() {
        for ch1 != nil || ch2 != nil {
            select {
            case v, open := <-ch1:
                if !open {
                    ch1 = nil
                    break
                }
            ch <- v
                case v, open := <-ch2:
                if !open {
                    ch2 = nil
                    break
                }
                ch <- v
            }
        }
        close(ch)
    }()
    return ch
}

closed channel nevel blocks

package main

import "fmt"

func main() {
    ch := make(chan bool, 2)
    ch <- true
    ch <- true
    close(ch)

    for i := 0; i < cap(ch) +1 ; i++ {
    v, ok := <- ch
    fmt.Println(v, ok)
    }
}

// 返回
// true true
// true true
// false false  第一个 false 是指在 channel 中的 value 类型的零值，也就是 false，第二个 false 是指 channel 的状态，

如果 for 循环中的条件改成 i < 100，那么循环 100 次。正确的感知 channel 被 close 掉，使用 range

package main

import "fmt"

func main() {
    ch := make(chan bool, 2)
    ch <- true
    ch <- true
    close(ch)

    for v := range ch {
    fmt.Println(v) // called twice
    }
}

A receive operation on a closed channel can always proceed immediately, yielding the element type's zero value after any previously sent values have been received.
close is effectively a broadcast signal to the senders，类似于广播的概念

利用 close(chan) 来关闭

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    finish := make(chan struct{})
    var done sync.WaitGroup
    done.Add(1)
    go func() {
        select {
        case <-time.After(1 * time.Hour):
        case <-finish:
        }
        done.Done()
    }()
    t0 := time.Now()
    close(finish)
    done.Wait()
    fmt.Printf("Waited %v for goroutine to stop\n", time.Since(t0))
}

利用 channel 能始终接收 close 的信号，在 channel 作为信号的时候，能 close(finish)，这样避免忘记发送信号。

A nil channel always block

一个没有初始化的 channel 会一直阻塞

package main

func main() {
    var ch chan bool
    ch <- true // blocks forever
}

因为没有初始化 channel，一个 nil channel 则会阻塞，同时等待多个 channel

// WaitMany waits for a and b to close.
func WaitMany(a, b chan bool) {
    var aclosed, bclosed bool
    for !aclosed || !bclosed {
    select {
    case <-a:
        aclosed = true
    case <-b:
        bclosed = true
    }
    }
}

即使 a close 后，a channel 仍然能够不停的读，在陷入到了 for 的死循环中，
而让 channel b 永远不会被判定为关闭。

具体可见这里的输出

package main

import (
    "fmt"
    "time"
)

func WaitMany(a, b chan bool) {
    for a != nil || b != nil {
    select {
    case <-a:
        a = nil
    case <-b:
        b = nil
    }
    }
}

func main() {
        a, b := make(chan bool), make(chan bool)
        t0 := time.Now()
        go func() {
                close(a)
                close(b)
        }()
        WaitMany(a, b)
        fmt.Printf("waited %v for WaitMany\n", time.Since(t0))
}

这个地方，使用 nil channel 之后，当 nil channel 在 select 中，会被优化掉，因为 nil channel 永远是读不了的，
leaving only b which blocks until it is closed, exiting the loop without spinning.

优雅退出的实现

// not gracefully
package main

import (
    "fmt"
    "time"
)

type Task struct {
    ticker *time.Ticker
}

func (t *Task) Run() {
    for {
        select {
        case <-t.ticker.C:
            handle()
        }
    }
}

func handle() {
    for i := 0; i < 5; i++ {
        fmt.Print("#")
        time.Sleep(time.Millisecond * 200)
    }
    fmt.Println()
}

func main() {
    task := &Task{
        ticker: time.NewTicker(time.Second * 2),
    }
    task.Run()
}
//
// $ go run main.go
// #####
// ###^Csignal: interrupt

// Method 1: use goroutine to exit gracefully
func main() {
    task := &Task{
        ticker: time.NewTicker(time.Second * 2),
    }

    c := make(chan os.Signal)
    signal.Notify(c, os.Interrupt)

    go func() {
        select {
        case sig := <-c:
            fmt.Printf("Got %s signal. Aborting...\n", sig)
            os.Exit(1)
        }
    }()

    task.Run()
}

// Another way but more graceful
// Add a channel to Task
// It works becacuse the task is running in the main goroutine
// So what if the task is running in an ordinary goroutine???
//
type Task struct {
    closed chan struct{}
    ticker *time.Ticker
}

func (t *Task) Run() {
    for {
        select {
        // If we receive a signal, then we exits
        case <-t.closed:
            return
        case <-t.ticker.C:
            handle()
        }
    }
}

// A wrapper to notify
func (t *Task) Stop() {
    close(t.closed)
}

func main() {
    task := &Task{
        ticker: time.NewTicker(time.Second * 2),
    }

    c := make(chan os.Signal)
    // Capcture the Ctrl + C signal
    signal.Notify(c, os.Interrupt)
    // a go
    go func() {
        select {
        case sig := <-c:
            fmt.Printf("Got %s signal. Aborting...\n", sig)
            // notify another go routine
            task.Stop()
        }
    }()

    task.Run()

}

// main goroutine waits a goroutine to finish
// use sync.WaitGroup
// a new definition to tasks
type Task struct {
    closed chan struct{}
    wg     sync.WaitGroup  //
    ticker *time.Ticker
}

func main() {
    // previous code...
    task.wg.Add(1)
    // in a goroutine got do tasks
    go func() { defer task.wg.Done(); task.Run() }()

    // other code...
}

// the whole program
package main

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

type Task struct {
    closed chan struct{}
    wg     sync.WaitGroup
    ticker *time.Ticker
}

func (t *Task) Run() {
    for {
        select {
        case <-t.closed:
            return
        case <-t.ticker.C:
            handle()
        }
    }
}

func (t *Task) Stop() {
    close(t.closed)
    t.wg.Wait()
}

func handle() {
    for i := 0; i < 5; i++ {
        fmt.Print("#")
        time.Sleep(time.Millisecond * 200)
    }
    fmt.Println()
}

func main() {
    task := &Task{
        closed: make(chan struct{}),
        ticker: time.NewTicker(time.Second * 2),
    }

    c := make(chan os.Signal)
    signal.Notify(c, os.Interrupt)

    task.wg.Add(1)
    go func() { defer task.wg.Done(); task.Run() }()

    select {
    case sig := <-c:
        fmt.Printf("Got %s signal. Aborting...\n", sig)
        task.Stop()
    }
}

goroutine 的退出机制：只能由本身控制，不允许从外部强制结束该 goroutine。
只有两种情况例外，那就是 main 函数结束或者程序崩溃结束运行。

几种并发控制的形式

使用 WaitGroup

通过 Add(int) 方法在每次 go func 之前增加计数，并在 goroutine 中使用 Done() 方法使计数减 1，

在主进程中通过调用 Wait() 方法等待所有 goroutine 执行完毕，再执行之后的逻辑。

package main

import (
    "sync"
    "fmt"
)

func main() {
    var wg sync.WaitGroup

    for i := 0; i < 10; i ++ {
    wg.Add(1)
    go func(i int) {
        defer func() {
        wg.Done()
        }()
        fmt.Printf("Go func %d\n", i)
    }(i)
    }

    wg.Wait()
    fmt.Println("All work done.")
}

问题：利用goroutine 执行完成后这个工作才真正完成。
但是如果中间超时或者 goroutine 任务在某个地方不断循环，就会导致主线程无限等待下去，因此需要一种机制来更主动地监控、控制 goroutine 的运行。

channel 的使用

func (s *Scheduler) Start() {
    go func() {
        s.ticker = time.NewTimer(s.interval)
    for {
        select {
        case <-s.interrupt :
                    // 在此做一些退出前的必要操作
            return
                case <-s.ticker.C:
                    // 在此循环运行任务
    }
    }()
}

func (s *Scheduler) Shutdown() {
    /* 中断调度 */
    s.interrupt <- true
}

使用 context

// Stream generates values with DoSomething and sends them to out
 // until DoSomething returns an error or ctx.Done is closed.
func Stream(ctx context.Context, out chan<- Value) error {
   for {
    v, err := DoSomething(ctx)
    if err != nil {
        return err
    }
    select {
        case <-ctx.Done():
        return ctx.Err()
        case out <- v:
    }
   }
}

官方提供了两个 Context，他们都是空 context，不可取消、无截止时间、没有任何携带值：
- context.Background() 返回一个 context，一般用于根节点
- context.TODO() 返回一个 context，当不知道应该用什么 Context 的时候，一般使用这个

下面方法，能分别获得相应字段衍生的 Context：

context.WithCancel(parent Context) (ctx Context, cancel CancelFunc)

context.WithDeadline(parent Context, d time.Time) (Context, CancelFunc)

context.WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc)

对于带有返回的 CancelFunc 的 context，能在父 goroutine 中通过调用返回的 CancelFunc 来实现主动通知子 goroutine 退出，

子 context 也会相应收到通知退出。

    context.WithValue(parent Context, key, val interface{}) Context

对于 WithValue 方法获得的 context，能通过 context.Value(key) 方法获取相应的数据。
但是对于关键的数据最好不要通过 context 来传递。

一个例子

package main

import (
  "context"
  "fmt"
  "math/rand"
  "time"
)

// Slow function
func sleepRandom(fromFunction string, ch chan int) {
  // defer cleanup
  defer func() { fmt.Println(fromFunction, "sleepRandom complete") }()

  //Perform a slow task
  //For illustration purpose,
  //Sleep here for random ms
  seed := time.Now().UnixNano()
  r := rand.New(rand.NewSource(seed))
  randomNumber := r.Intn(100)
  sleeptime := randomNumber + 100
  fmt.Println(fromFunction, "Starting sleep for", sleeptime, "ms")
  time.Sleep(time.Duration(sleeptime) * time.Millisecond)
  fmt.Println(fromFunction, "Waking up, slept for ", sleeptime, "ms")

  // write on the channel if it was passed in
  if ch != nil {
    ch <- sleeptime
  }
}

//Function that does slow processing with a context
//Note that context is the first argument
func sleepRandomContext(ctx context.Context, ch chan bool) {

  //Cleanup tasks
  //There are no contexts being created here
  //Hence, no canceling needed
  defer func() {
    fmt.Println("sleepRandomContext complete")
    ch <- true
  }()

  //Make a channel
  sleeptimeChan := make(chan int)

  //Start slow processing in a goroutine
  //Send a channel for communication
  go sleepRandom("sleepRandomContext", sleeptimeChan)

  //Use a select statement to exit out if context expires
  select {
  case <-ctx.Done():
    //If context is cancelled, this case is selected
    //This can happen if the timeout doWorkContext expires or
    //doWorkContext calls cancelFunction or main calls cancelFunction
    //Free up resources that may no longer be needed because of aborting the work
    //Signal all the goroutines that should stop work (use channels)
    //Usually, you would send something on channel,
    //wait for goroutines to exit and then return
    //Or, use wait groups instead of channels for synchronization
    fmt.Println("sleepRandomContext: Time to return")
  case sleeptime := <-sleeptimeChan:
    //This case is selected when processing finishes before the context is cancelled
    fmt.Println("Slept for ", sleeptime, "ms")
  }
}

//A helper function, this can, in the real world do various things.
//In this example, it is just calling one function.
//Here, this could have just lived in main
func doWorkContext(ctx context.Context) {

  //Derive a timeout context from context with cancel
  //Timeout in 150 ms
  //All the contexts derived from this will returns in 150 ms
  ctxWithTimeout, cancelFunction := context.WithTimeout(ctx, time.Duration(150)*time.Millisecond)

  //Cancel to release resources once the function is complete
  defer func() {
    fmt.Println("doWorkContext complete")
    cancelFunction()
  }()

  //Make channel and call context function
  //Can use wait groups as well for this particular case
  //As we do not use the return value sent on channel
  ch := make(chan bool)
  go sleepRandomContext(ctxWithTimeout, ch)

  //Use a select statement to exit out if context expires
  select {
  case <-ctx.Done():
    //This case is selected when the passed in context notifies to stop work
    //In this example, it will be notified when main calls cancelFunction
    fmt.Println("doWorkContext: Time to return")
  case <-ch:
    //This case is selected when processing finishes before the context is cancelled
    fmt.Println("sleepRandomContext returned")
  }
}

func main() {
  //Make a background context
  ctx := context.Background()
  //Derive a context with cancel
  ctxWithCancel, cancelFunction := context.WithCancel(ctx)

  //defer canceling so that all the resources are freed up
  //For this and the derived contexts
  defer func() {
    fmt.Println("Main Defer: canceling context")
    cancelFunction()
  }()

  //Cancel context after a random time
  //This cancels the request after a random timeout
  //If this happens, all the contexts derived from this should return
  go func() {
    sleepRandom("Main", nil)
    cancelFunction()
    fmt.Println("Main Sleep complete. canceling context")
  }()
  //Do work
  doWorkContext(ctxWithCancel)
}

BruceChen7 / gitblog

goroutine使用 #43

资料