golang学习随便记9

Posted 2021-11-23 sjg20010414

goroutine和channel  （1）

golang有两种并发编程风格，基于CSP的goroutine+channel或者传统的共享内存多线程模型。毫无疑问，goroutine和channel是golang引以为傲的东西，虽然背后理论不是革命性的东西，但它足够有效率，所以成为golang立身之本。

goroutine

每一个并发执行的活动称为goroutine，它是一个逻辑概念，因为程序员不清楚一个goroutine是否会对应一个线程或纤程（goroutine不可能是线程，因为靠线程是不可能支持成千上万的并发的），但理解上，依然可以把它当线程看待。

一个程序启动时，只有一个goroutine来调用main函数，这个称为main goroutine。新的goroutine通过 go 语句创建，语法上，一个 go 语句就是在普通的函数或者方法调用前加上 go关键字 作为前缀。go语句将使函数在新创建的 goroutine 中调用，而 go语句 本身的执行是立刻完成的。

package main

import (
	"fmt"
	"time"
)

func main() {
	go spinner(100 * time.Millisecond)            // 创建 goroutine 后立刻继续往下执行
	const n = 45
	fibN := fib(n)
	fmt.Printf("\\rFibonacci(%d) = %d\\n", n, fibN)
}

func spinner(delay time.Duration) {
	for {
		for _, r := range `-\\|/` {
			fmt.Printf("\\r%c", r)
			time.Sleep(delay)
		}
	}
}

func fib(x int) int {
	if x < 2 {
		return x
	}
	return fib(x-1) + fib(x-2)
}

上面的代码中，用 字符旋转效果 来表示程序在运行中，这个 goroutine 执行不会影响 main goroutine 的执行，事实上，spinner一旦执行（在新的goroutine）就往下去计算fibonacci数列n=45时的值了。spinner是一个“死循环”，所以，它对应的goroutine只能等main返回或程序终结才会被终结。

func main() {
	listener, err := net.Listen("tcp", "localhost:8000")
	if err != nil {
		log.Fatal(err)
	}
	for {
		conn, err := listener.Accept()
		if err != nil {
			log.Print(err)
			continue
		}
		handleConn(conn)
	}
}

func handleConn(c net.Conn) {
	defer c.Close()
	for {
		_, err := io.WriteString(c, time.Now().Format("15:04:05\\n"))
		if err != nil {
			return
		}
		time.Sleep(1 * time.Second)
	}
}

编译上面的程序，运行，然后用putty，选择Raw方式连接到8000端口，就发现每隔一秒打印一下时间。程序中main内的for是一个“死循环”，即永远等待连接。listener.Accept() 会阻塞，直到有连接请求进来，然后返回 net.Conn 对象表示这个连接，用 handleConn 函数处理这个连接（永远打印时间）。显然，这个程序只支持一个连接（因为处理连接的handleConn是死循环，占满了唯一的goroutine），我们可以开两个putty窗口来验证。更准确地，我们可以依次打开1、2、3三个putty窗口，会发现只有1窗口有时间输出，此时，关闭窗口1，很快窗口2有时间出现了，同样，再关闭窗口2，窗口3会有时间出现。

 没有putty，可以像书上那样直接写一个类似 netcat 命令的程序。

要让这个程序支持多个连接，只需要很小改动，就是处理连接的函数 handleConn 在自己的 goroutine中运行（改成 go handleConn(conn)），这样程序就可以立刻等待下一个连接的到来。

下面的例子，服务端和上面的程序差不多，监听端口，有连接了，在独立的 goroutine中处理这个连接，然后继续监听，不会阻塞。处理过程就是将客户端发来的串回声输出大写、原串、小写（输出目标是网络流）。客户端则连接到服务器之后，在独立的goroutine将网络流输出到标准输出，然后将标准输入“拷贝”到网络流发送。最终效果就是客户端从标准输入输入的东西，被服务器处理后传回客户端，再被客户端输出到标准输出。

服务端程序

func main() {
	listener, err := net.Listen("tcp", "localhost:8000")
	if err != nil {
		log.Fatal(err)
	}

	for {
		conn, err := listener.Accept()
		if err != nil {
			log.Print(err) // 连接终止
			continue
		}
		go handleConn(conn) // 在独立的 goroutine 处理连接狗go 
	}
}

func echo(c net.Conn, shout string, delay time.Duration) {
	fmt.Fprintln(c, "\\t", strings.ToUpper(shout))
	time.Sleep(delay)
	fmt.Fprintln(c, "\\t", shout)
	time.Sleep(delay)
	fmt.Fprintln(c, "\\t", strings.ToLower(shout))
}

func handleConn(c net.Conn) {
	input := bufio.NewScanner(c)
	for input.Scan() {
		echo(c, input.Text(), 1*time.Second)
	}
	c.Close()
}

客户端程序 netcat

func main() {
	conn, err := net.Dial("tcp", "localhost:8000")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	go mustCopy(os.Stdout, conn) // 在独立 goroutine 输出到终端
	mustCopy(conn, os.Stdin)
}

func mustCopy(dst io.Writer, src io.Reader) {
	if _, err := io.Copy(dst, src); err != nil {
		log.Fatal(err)
	}
}

channel

用go关键字写出不会阻塞的并发程序显然不是golang并发的全部，因为并发的程序相互之间不扯上关系的程序很少。之前写Visual C++多线程就体会到多线程非常难调试，而且容易出错而找不到位置，在传统中，线程间交流信息就是共享一块内存，然后约定一些维护它的机制（我之前用的是关键段Critical Section）。golang有传统共享内存机制，但它还有 channel通信机制。

每个channel都有一个特殊的类型，也就是channels可发送数据的类型。一个可以发送int类型数据的channel一般写为chan int。ch := make(chan int) // ch has type 'chan int'

channel类型的变量是对底层数据结构的引用，因此，复制channel或者作为函数参数传递，是引用拷贝，指向同一个底层数据的两个channel变量，用 == 比较时为 true （channel类型零值为nil）。

channel的操作主要是两个：发送和接收，关键字符号都是 <-  （->已经被指针用掉了，只能都用<-了，哈哈），靠位置来区分是发送还是接收

ch <- x   // 用 ch 发送 x
x = <-ch  // 从 ch 取出值赋值给 x
<-ch      // 从 ch 取出值，但扔掉不用

channel还可以进行关闭操作，方法是用内置函数close：close(ch)。可以认为，用内置 make 创建一个 channel 就是“打开”，而所谓用close关闭channel，意思是不允许再往channel里面“灌”数据（再“灌”panic），被关闭的channel，接收方仍然可以接收（还有没有接收完的数据，就是继续接收，没有未接收的数据，就是接收到零值）。

不带缓存的channel

无缓存的channel是阻塞的：一个基于无缓存Channels的发送操作将导致发送者goroutine阻塞，直到另一个goroutine在相同的Channels上执行接收操作，当发送的值通过Channels成功传输之后，两个goroutine可以继续执行后面的语句。反之，如果接收操作先发生，那么接收者goroutine也将阻塞，直到有另一个goroutine在相同的Channels上执行发送操作。

无缓存channel的发送和接收操作将导致两个 goroutine 做一次同步操作，因此，无缓存channel也常常称为同步 channel，或者说实现同步是无缓存 channel 的用法。无缓存channel发送数据时，发送者goroutine会“挂起”，只有接收者收到数据了，才可能会再次唤醒发送者 goroutine（具体是立刻唤醒还是滞后唤醒不是我们操心的，也不能控制）。

在前面的客户端程序 netcat 中，如果 main goroutine 输入 Ctrl+Z 或 Ctrl+D 结束了输入，后台 goroutine 可能还在干活，这样就有可能服务器发回的信息没有来得及显示程序就结束了。我们可以用同步 channel 来确保服务器返回的信息得到显示。

    done := make(chan struct{})
    go func() {
        io.Copy(os.Stdout, conn) // NOTE: ignoring errors
        log.Println("done")
        done <- struct{}{} // signal the main goroutine
    }()
    mustCopy(conn, os.Stdin)
    conn.Close()
    <-done // wait for background goroutine to finish

同步 channel 因为只是为了传递一个“信号”，所以，常常发送一个 struct{}  （表示不是什么具体类型）。另外，常常用即时调用的匿名函数包裹需要同步的 goroutine。

串联的 channels （管道 pipeline）

使用多个发送具体值的同步 channel，并串联起来，就可以形成管道。

package main

import (
	"fmt"
	"time"
)

func main() {
	naturals := make(chan int)
	squares := make(chan int)

	go func() {
		for x := 0; ; x++ {
			naturals <- x // 用 naturals 发送 x
			time.Sleep(500 * time.Millisecond)
		}
	}()

	go func() {
		for {
			x := <-naturals  // 从 naturals 获得 x
			squares <- x * x // 用 squares 发送 x*x
		}
	}()

	for {
		fmt.Println(<-squares) // 打印的是从 squares 获得的值
	}
}

上面的程序，将会产生无穷的数列，有些场合可能适合这种一运行就不停工作在某种工作流中。如果我们希望接收者知道发送者不再发送了，那么可以使用内置close函数来关闭一个 channel。

close(naturalse)

当一个 channel 被关闭后，就不能再向里面“灌”数据了（panic），但是对于接收者，仍然可以从已经关闭的 channel 中接收数据——如果还有数据，就正常接收，如果没有数据了，会得到零值，并且接收操作不再是阻塞的。修改成下述代码

        for x := 0; ; x++ {
			if x >= 5 {
				close(naturals)
				break
			}
			naturals <- x // 用 naturals 发送 x
			time.Sleep(500 * time.Millisecond)
		}

输出时，先每隔半秒输出0、1、4、9、16，然后快速不停输出 0

不能直接测试一个 channel 是否是关闭的，但可以让接收者多接收一个布尔值ok，表示是否成功从 channel 接收到了值。

    go func() {
		for {
			x, ok := <-naturals // 从 naturals 获得 x
			if !ok {
				break
			}
			squares <- x * x // 用 squares 发送 x*x
		}
		close(squares)
	}()

	for {
		if y, ok := <-squares; ok {
			fmt.Println(y) // 打印的是从 squares 获得的值
		}
        break
	}

因为经常需要处理 channel 中没有值结束循环的问题，所以，golang 有语法糖 range 循环：range 在 channel 上迭代，它依次从 channel 接收数据，当 channel 被关闭并且没有值可以被接收时跳出循环。

channel 并非文件那样的资源，所以，不是说每个 channel 必须关闭，如果一个 channel 不再被引用，golang的GC会自动回收它。

单方向的 channel

上面的代码中，产生自然数和平方这两个goroutine都是匿名函数形式嵌入在main goroutine 中，实际工程中，我们总是希望将功能拆分到独立的函数，这时，就要考虑 channel 是放到全局的变量还是作为函数的参数，显然，后者好一点。channel 一旦作为函数的参数，就从双向变成单向的（即作为函数参数的channel只能一个方向，channel变量传入函数，发生双向变单向的隐式转换），作为函数参数的 channel 需要区分负责接收（类型 <-chan  T）还是发送的（类型 chan<- T），并且，通常用变量命名来明确。

package main

import (
	"fmt"
	"time"
)

func counter(out chan<- int) {
	for x := 0; x < 5; x++ {
		out <- x
		time.Sleep(1 * time.Second)
	}
	close(out)
}

func squarer(out chan<- int, in <-chan int) {
	for v := range in {
		out <- v * v
	}
	close(out)
}

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

func main() {
	naturals := make(chan int)
	squares := make(chan int)

	go counter(naturals)
	go squarer(squares, naturals)
	printer(squares)
}

带缓存的 channels

带缓存的 channel 也就是内部维护一个元素队列的 channel，创建的方法是 用make函数第二个参数指定队列最大容量（内置函数 cap(ch) 和 len(ch) 可以获得 channel内部队列的容量和有效元素个数），如 ch = make(chan string, 3)

向缓存Channel的发送操作就是向内部缓存队列的尾部插入元素，接收操作则是从队列的头部删除元素。如果内部缓存队列是满的，那么发送操作将阻塞直到因另一个goroutine执行接收操作而释放了新的队列空间。相反，如果channel是空的，接收操作将阻塞直到有另一个goroutine执行发送操作而向队列插入元素。队列不满也不空，那么对于发送和接收都不会阻塞。

书上举的例子是3个goroutine向不同的URL请求，然后将响应最快的那个结果返回——这个操作对于通常的编程并不容易。

func mirroredQuery() string {
    responses := make(chan string, 3)
    go func() { responses <- request("asia.gopl.io") }()
    go func() { responses <- request("europe.gopl.io") }()
    go func() { responses <- request("americas.gopl.io") }()
    return <-responses // return the quickest response
}

func request(hostname string) (response string) { /* ... */ }

每个 go func() 都将请求的响应通过 channel 发送， mirroredQuery 返回的是接收下来的值（最先发送的数据，也就是最先响应的URL的数据，它被接收并返回，另外2个值也会被放入队列，因为队列容量确保了3个值都能被容纳，无论有无接收者，这里另外2个值没有接收者）。如果使用无缓存的 channel，那么慢的两个响应值因为没有接收者，就会永远卡住，这种情况称为 goroutine 泄漏，无法被 GC 回收，因为，必须确保不再需要的 goroutine 正常退出（不能有僵尸 goroutine）。

上面的例子中，发送的值总是能被放入队列，但总是分配 channel 缓存为可能发送的元素的数目是不现实的，但具体 channel 缓存大小多少合适，跟具体问题有关。