云原生训练营模块二 Go语言进阶

Posted 2022-05-06 果子哥丶

这里写目录标题

• 函数
• • Main函数
  • init函数
  • 传递变长参数
  • 内置函数
  • 回调函数
  • 闭包
  • 接口
  • 反射机制
• 常用语法
• • 错误处理
  • defer
  • Panic和recover
• 多线程
• channel - 多线程通信
• • 通道缓冲
  • 遍历通道缓冲区
• 关闭通道
• • Select
  • 定时器Timer
  • 上下文Context
  • 如何停止一个子协程
• 线程加锁
• 线程调度
• Goroutine
• 课后练习

• 函数调用
• 常用语法
• 多线程
• 深入理解channel
• 基于channel编写一个生产者消费者程序

函数

Main函数

• 每个 Go 语言程序都应该有个 main package
• Main package 里的 main 函数是 Go 语言程序入口

package main
func main() 
	args := os.Args
	if len(args) != 0 
		println("Do not accept any argument")
		os.Exit(1)
	
	println("Hello world")

init函数

• Init 函数：会在包初始化时运行
• 谨慎使用 init 函数
  • 当多个依赖项目引用统一项目，且被引用项目的初始化在 init 中完成，并且不可重复运行时，会导
    致启动错误

package main
var myVariable = 0
func init() 
	myVariable = 1

传递变长参数

Go 语言中的可变长参数允许调用方传递任意多个相同类型的参数

• 函数定义

func append(slice []Type, elems ...Type) []Type

• 调用方法

myArray := []string
myArray = append(myArray, "a","b","c")

内置函数

回调函数

函数作为参数传入其它函数，并在其他函数内部调用执行

• strings.IndexFunc(line, unicode.IsSpace)
• Kubernetes controller的leaderelection

示例：

func main() 
	DoOperation(1, increase)
	DoOperation(1, decrease)

func increase(a, b int) 
	println(“increase result is:”, a+b)

func DoOperation(y int, f func(int, int)) 
	f(y, 1)

func decrease(a, b int) 
	println("decrease result is:", a-b)

闭包

匿名函数

• 不能独立存在
• 可以赋值给其他变量
  x:= func()
• 可以直接调用
  func(x,y int)println(x+y)(1,2)
• 可作为函数返回值
  func Add() (func(b int) int
  

接口

• 接口定义一组方法集合

  type IF interface 
  	Method1(param_list) return_type
  
  

• 适用场景：Kubernetes 中有大量的接口抽象和多种实现

• Struct 无需显示声明实现 interface，只需直接实现方法

• Struct 除实现 interface 定义的接口外，还可以有额外的方法

• 一个类型可实现多个接口（Go 语言的多重继承）

• Go 语言中接口不接受属性定义

• 接口可以嵌套其他接口

• 注意事项

  • Interface 是可能为 nil 的，所以针对 interface 的使用一定要预
    先判空，否则会引起程序 crash(nil panic)
  • Struct 初始化意味着空间分配，对 struct 的引用不会出现空指针

反射机制

• reflect.TypeOf ()返回被检查对象的类型
• reflect.ValueOf()返回被检查对象的值

myMap := make(map[string]string, 10)
myMap["a"] = "b"
t := reflect.TypeOf(myMap)
fmt.Println("type:", t)
v := reflect.ValueOf(myMap)
fmt.Println("value:", v)

常用语法

错误处理

• Go 语言无内置 exception 机制，只提供 error 接口供定义错误

  type error interface 
  Error() string
  
  

• 可通过 errors.New 或 fmt.Errorf 创建新的 error
  var errNotFound error = errors.New("NotFound")

• 通常应用程序对 error 的处理大部分是判断 error 是否为 nil，
  如需将 error 归类，通常交给应用程序自定义，比如 kubernetes 自定义了与 apiserver 交互的不同类型错误。

type StatusError struct 
	ErrStatus metav1.Status

var _ error = &StatusError

// Error implements the Error interface.
func (e *StatusError) Error() string 
	return e.ErrStatus.Message

defer

函数返回之前执行某个语句或函数

• 等同于 Java 和 C# 的 finally

常见的 defer 使用场景：记得关闭你打开的资源

• defer file.Close()
• defer mu.Unlock()
• defer println(“”)

Panic和recover

• panic: 可在系统出现不可恢复错误时主动调用 panic, panic 会使当前线程直接 crash
• defer: 保证执行并把控制权交还给接收到 panic 的函数调用者
• recover: 函数从 panic 或 错误场景中恢复

defer func() 
	fmt.Println("defer func is called")
	if err := recover(); err != nil 
		fmt.Println(err)
	
	()
panic("a panic is triggered")

多线程

• 进程：

  • 分配系统资源（CPU 时间、内存等）基本单位
  • 有独立的内存空间，切换开销大

• 线程：进程的一个执行流，是 CPU 调度并能独立运行的的基本单位

  • 同一进程中的多线程共享内存空间，线程切换代价小
  • 多线程通信方便
  • 从内核层面来看线程其实也是一种特殊的进程，它跟父进程共享了打开的文件和文件系统信息，共
    享了地址空间和信号处理函数

• 协程

  • Go 语言中的轻量级线程实现
  • Golang 在 runtime、系统调用等多方面对 goroutine 调度进行了封装和处理，当遇到长时间执行
    或者进行系统调用时，会主动把当前 goroutine 的 CPU § 转让出去，让其他 goroutine 能被调度
    并执行，也就是 Golang 从语言层面支持了协程

CSP
描述两个独立的并发实体通过共享的通讯 channel 进行通信的并发模型。

Go 协程 goroutine

• 是一种轻量线程，它不是操作系统的线程，而是将一个操作系统线程分段使用，通过调度器实现协作式调度。
• 是一种绿色线程，微线程，它与 Coroutine 协程也有区别，能够在发现堵塞后启动新的微线程。

通道 channel

• 类似 Unix 的 Pipe，用于协程之间通讯和同步。
• 协程之间虽然解耦，但是它们和 Channel 有着耦合。

线程和协程的差异

• 每个 goroutine (协程) 默认占用内存远比 Java 、C 的线程少

  • goroutine：2KB
  • 线程：8MB

• 线程/goroutine 切换开销方面，goroutine 远比线程小

  • 线程：涉及模式切换(从用户态切换到内核态)、16个寄存器、PC、SP…等寄存器的刷新
  • goroutine：只有三个寄存器的值修改 - PC / SP / DX.

• GOMAXPROCS

  • 控制并行线程数量

启动新协程：go functionName()

for i := 0; i < 10; i++ 
	go fmt.Println(i)

time.Sleep(time.Second)

channel - 多线程通信

Channel 是多个协程之间通讯的管道

• 一端发送数据，一端接收数据
• 同一时间只有一个协程可以访问数据，无共享内存模式可能出现的内存竞争
• 协调协程的执行顺序

声明方式

• var identifier chan datatype
• 操作符<-

示例

ch := make(chan int)
go func() 
	fmt.Println("hello from goroutine")
	ch <- 0 //数据写入Channel
()
i := <-ch //从Channel中取数据并赋值

通道缓冲

• 基于 Channel 的通信是同步的
• 当缓冲区满时，数据的发送是阻塞的
• 通过 make 关键字创建通道时可定义缓冲区容量，默认缓冲区容量为 0

遍历通道缓冲区

ch := make(chan int, 10)
go func() 
	for i := 0; i < 10; i++ 
	rand.Seed(time.Now().UnixNano())
	n := rand.Intn(10) // n will be between 0 and 10
	fmt.Println("putting: ", n)
	ch <- n
	
	close(ch)
()
fmt.Println("hello from main")
for v := range ch 
	fmt.Println("receiving: ", v)

关闭通道

• 通道无需每次关闭
• 关闭的作用是告诉接收者该通道再无新数据发送
• 只有发送方需要关闭通道

ch := make(chan int)
defer close(ch)
if v, notClosed := <-ch; notClosed 
	fmt.Println(v)

Select

定时器Timer

上下文Context

如何停止一个子协程

基于 Context 停止子协程

• Context 是 Go 语言对 go routine 和 timer 的封装
  

线程加锁

理解线程安全

锁：

• Go 语言保证线程安全，可以使用 channel 和 共享内存去保证。
• Go 语言不仅仅提供基于 CSP 的通信模型，也支持基于共享内存的多线程数据访问，在Sync包提供了锁的基本原语。
• sync.Mutex 互斥锁，Lock加锁，unlock解锁。不论读和写都是互斥的。
• sync.RWMutex 读写分离锁，不限制并发读，只限制并发写和并发读写。
• sync.WaitGroup 它的语意就是定义一个组，这个组里面会有假如100个线程，每个线程在结束时候都应该去调Done()，只有结束Done()减为0的时候才往下执行wait()
• sync.Once 保证某段代码只执行一次
• sync.Cond 让一组Goroutine 在满足特定条件时被唤醒（生产者、消费者）

sync.NewCond(&sync.Mutex)

package main

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

func main() 
	defer fmt.Println("1")
	defer fmt.Println("2")
	defer fmt.Println("3")
	loopFunc()
	time.Sleep(time.Second)


func loopFunc() 
	lock := sync.Mutex
	for i := 0; i < 3; i++ 
		// go func(i int) 
		lock.Lock()
		defer lock.Unlock()
		fmt.Println("loopFunc:", i)
		// (i)
	

mutex示例：

package main

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

func main() 
	go rLock()
	go wLock()
	go lock()
	time.Sleep(5 * time.Second)


func lock() 
	lock := sync.Mutex
	for i := 0; i < 3; i++ 
		lock.Lock()
		defer lock.Unlock()
		fmt.Println("lock:", i)
	


func rLock() 
	lock := sync.RWMutex
	for i := 0; i < 3; i++ 
		lock.RLock()
		defer lock.RUnlock()
		fmt.Println("rLock:", i)
	


func wLock() 
	lock := sync.RWMutex
	for i := 0; i < 3; i++ 
		lock.Lock()
		defer lock.Unlock()
		fmt.Println("wLock:", i)
	

cond示例：生产者与消费者

package main

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

type Queue struct 
	queue []string
	cond  *sync.Cond


func main() 
	q := Queue
		queue: []string,
		cond:  sync.NewCond(&sync.Mutex),
	
	go func() 
		for 
			q.Enqueue("a")
			time.Sleep(time.Second * 2)
		
	()
	for 
		q.Dequeue()
		time.Sleep(time.Second)
	


func (q *Queue) Enqueue(item string) 
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	q.queue = append(q.queue, item)
	fmt.Printf("putting %s to queue, notify all\\n", item)
	q.cond.Broadcast()


func (q *Queue) Dequeue() string 
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	for len(q.queue) == 0 
		fmt.Println("no data available, wait")
		q.cond.Wait()
	
	result := q.queue[0]
	q.queue = q.queue[1:]
	return result

线程调度

• 进程：资源分配的基本单位

进程切换开销

• 直接开销
  - 切换页表全局目录（PGD）
  - 切换内核态堆栈
  - 切换硬件上下文（进程恢复前，必须装入寄存器的数据统称为硬件上下文）
  - 刷新TLB
  - 系统调度器的代码执行

• 间接开销
  - CPU缓存失效导致的进程需要到内存直接访问的IO操作变多

• 线程：调度的基本单位

线程切换开销

• 线程本质上只是一批共享资源的进程，线程切换本质上依然需要内核进行进程切换
• 一组线程因为共享内存资源，因此一个进程的所有线程共享虚拟地址空间，线程切换相比进程切换，主要节省了虚拟地址空间的切换

• 无论是线程还是进程，在linux中都以task_strut描述，从内核角度看，与进程无本质区别。
• Glibc中的pthread库提供NPTL（Native POSIX Threading Library）支持
  

用户线程：无需内核帮助，应用程序在用户空间创建的可执行单元，创建销毁完全在用户态完成，减少内核态（系统调用的依赖）的消耗。

Goroutine

Go 语言基于 GMP 模型实现用户态线程

• G：表示 goroutine，每个 goroutine 都有自己的栈空间，定时器，
  初始化的栈空间在 2k 左右，空间会随着需求增长。
• M（相当于CPU数）：抽象化代表内核线程，记录内核线程栈信息，当 goroutine 调度
  到线程时，使用该 goroutine 自己的栈信息。
• P：代表调度器，负责调度 goroutine，维护一个本地 goroutine 队
  列，M 从 P 上获得 goroutine 并执行，同时还负责部分内存的管理。

G所处的位置

• 进程都有一个全局的 G 队列
• 每个 P 拥有自己的本地执行队列
• 有不在运行队列中的 G
  • 处于 channel 阻塞态的 G 被放在 sudog
  • 脱离 P 绑定在 M 上的 G，如系统调用
  • 为了复用，执行结束进入 P 的 gFree 列表中的 G

Goroutine 创建过程

• 获取或者创建新的 Goroutine 结构体
  • 从处理器的 gFree 列表中查找空闲的 Goroutine
  • 如果不存在空闲的 Goroutine，会通过 runtime.malg 创建一个栈大小足够的新结构体
• 将函数传入的参数移到 Goroutine 的栈上
• 更新 Goroutine 调度相关的属性，更新状态为_Grunnable
• 返回的 Goroutine 会存储到全局变量 allgs 中

将 Goroutine 放到运行队列上

• Goroutine 设置到处理器的 runnext 作为下一个处理器执行的任务
• 当处理器的本地运行队列已经没有剩余空间时（256），就会把本地队列中的一部分 Goroutine 和待加入的 Goroutine通过 runtime.runqputslow 添加到调度器持有的全局运行队列上

调度器行为

• 为了保证公平，当全局运行队列中有待执行的 Goroutine 时，通过 schedtick 保证有一定
  几率（1/61）会从全局的运行队列中查找对应的 Goroutine
• 从处理器本地的运行队列中查找待执行的 Goroutine
• 如果前两种方法都没有找到 Goroutine，会通过 runtime.findrunnable 进行阻塞地查找
  Goroutine
  • 从本地运行队列、全局运行队列中查找
  • 从网络轮询器中查找是否有 Goroutine 等待运行
  • 通过 runtime.runqsteal 尝试从其他随机的处理器中窃取待运行的 Goroutine

课后练习

将练习1.2中的生产者消费者模型修改成为多个生产者和多个消费者模式

代码实现用了Cond条件变量和channel通道。

package main

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

var cond sync.Cond

//生产者
func produce(out chan<- int, nu int) 
	for 
		cond.L.Lock()
		//产品区满 等待消费者消费
		for len(out) == 3 
			cond.Wait()
		
		num := rand.Intn(1000)
		out <- num
		fmt.Printf("%dth ***producer produce***,num = %d,len(chan) = %d\\n", nu, num, len(out))
		cond.L.Unlock()

		//生产了产品唤醒 消费者线程
		cond.Signal()
		//生产完了歇一会，给其他协程机会
		time.Sleep(time.Second)
	


//消费者
func consume(in <-chan int, nu int) 
	for 
		cond.L.Lock()
		//产品区空 等待生产者生产
		for len(in) == 0 
			cond.Wait()
		
		num := <-in
		fmt.Printf("%dth ###consumer consume###,num = %d,len(chan) = %d\\n", nu, num, len(in))
		cond.L.Unlock()
		cond.Signal()

		//消费完了歇一会，给其他协程机会
		time.Sleep(time.Millisecond * 500)
	


func main() 
	//设置随机数种子
	rand.Seed(time.Now().UnixNano())
	quit := make(chan bool)
	//产品区 使用channel模拟
	product := make(chan int, 3)

	//创建互斥锁和条件变量
	cond.L = new(sync.Mutex)

	//5个消费者
	for i := 0; i < 5; i++ 
		go produce(product, i)
	
	//3个生产者
	for i := 0; i < 3; i++ 
		go consume(product, i)
	

	//主协程阻塞 不结束
	<-quit