命令记不住

登陆

Go 并发和 Channel

  |   129 浏览

Go 并发和 Channel

1.Go 并发原理

  • 并行与并发

    • 并发 Concurrency 逻辑上具备同时处理多个任务的能力
    • 并行 Parallelism 物理上在同一时刻执行多个并发任务

  • 并发 Goroutine

    • Go 的并发通过 Goroutine 实现，Goroutine 不等同于协程 Coroutine，而是两级线程模型，也称混合型线程模型
    • 主流线程模型
      • 内核级线程模型
      • 用户级线程模型 Coroutine
      • 两级线程模型 Goroutine
    • 线程默认的栈是 MB 级别， Goroutine 默认 2K

  • 使用 Goroutine

直接加go关键字

go func() {
	fmt.Println("Goroutine")
}()

package main

import (
	"fmt"
	"time"
)

func main() {

	go func() {
		fmt.Println("one")
	}()

	go func() {
		fmt.Println("two")
	}()
	go func() {
		fmt.Println("three")
	}()

	time.Sleep(time.Second * 1)
}

并发时钟服务器示例

服务端:

package main

import (
	"io"
	"log"
	"net"
	"time"
)

func main() {
	listener, err := net.Listen("tcp", "localhost:8000") //网络端口监听连接
	if err != nil {
		log.Fatal(err)
	}
	for {
		conn, err := listener.Accept() //accept方法被阻塞,直到有请求进来,返回代表一个连接
		if err != nil {
			log.Print(err)
			continue
		}
		go 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)
	}
}

客户端:

package main

import (
	"io"
	"log"
	"net"
	"os"
)

// 从网络连接中读取,然后写到标准输出,直到到达EOF或者出错
func main() {
	conn, err := net.Dial("tcp", "localhost:8000")
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	mustCopy(os.Stdout, conn)
}

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

2.Channel 简介

  • 并发 Channel

接受方向、数据类型、Buffer 大小

ch1 := make(chan int) //双向的channel可以用来收发，类型是int
//单向的channel主要用于函数传递的安全因素，类似于const一般不直接声明
ch2 := make(chan<- int)    //接受的channel可以用来收发，类型是int
ch3 := make(<-chan int)    //发送的channel可以用来收发，类型是int
ch4 := make(chan int, 100) //双向的channel可以用来收发，类型是int，大小是100
  • Channel 的使用
ch := make(chan int) // 双向的 channel ，类型是 int
ch <- 1 // 写入数据
a := <-ch // 读取数据
ch <- x //发送语句
x = <-ch //接收表达式
<-ch //接收语句
通道不再使用的时候可以调用 close 函数进行关闭
close(ch) // 关闭事件会通知到所有等在 ch 上的 goroutine， 不能关闭仅是接收方的ch

for i := range ch {
	// close 自动退出
	fmt.Println(i)
}
  • 配合 select 使用

select 用法类似与 IO 多路复用，可以同时监听多个 channel 的消息状态

select {
case <- ch1:
...
case <- ch2:
...
case ch3 <- 10;
...
default:
...
  • select 可以同时监听多个 channel 的写入或读取
  • 执行 select 时，若只有一个 case 通过(不阻塞)，则执行这个 case 块
  • 若有多个 case 通过，则随机挑选一个 case 执行
  • 若所有 case 均阻塞，且定义了 default 模块，则执行 default 模块。若未定义 default 模块，则 select 语句阻塞，直到有 case 被唤醒
  • 使用 break 会跳出 select 块。

多路复用

  package main
  
  import (
        "fmt"
        "runtime"
        "time"
  )
  
  func main(){
        //指定cpu核数,小于等于0全部使用
        fmt.Println("默认使用cpu核数:",runtime.GOMAXPROCS(0))
        runtime.GOMAXPROCS(1)
  
        intch := make(chan int,1)
        strch := make(chan string,1)
  
        intch <- 1
        strch <- "hello"
  
        /*
        1.select case如果未准备好，会阻塞
        2.如果多个case都满足条件会随机选择一个执行
        3.case必须为IO操作
        4.可以做超时和推出判断
        5.select 如果对应的是异步处理  需要在for循环中使用
        6.select可以监控chan数据流向
         */
        select {
        case value := <-intch:
                fmt.Println(value)
        case value := <-strch:
                fmt.Println(value)
        case <-time.After(time.Second * 5): //超时判断
                fmt.Println("超时")
                //default：阻塞操作
        }
  }
  C:\Users\vSphere\go\src\day04>go run 13.go
  默认使用cpu核数: 8
  1
  
  C:\Users\vSphere\go\src\day04>go run 13.go
  默认使用cpu核数: 8
  hello
  • Channel 小结

向已经 close 的 channel 写入数据会导致 panic

重复关闭 channel 或者 关闭为 nil 的 channel 也会引发 panic

如果 channel 为 nil，收发都会阻塞

同时监听多个 channel 或者 避免阻塞在 channel 上，可以配合 select 使用，select 在多个 channel 都满足条件的情况下，随机选择

一般原则是写入方负责关闭 channel，读取方负责检测 channel 是否关闭

3.channel 数据类型

  • 管道

管道可以用来连接goroutine,这样一个的输出是另一个的输入，这个叫管道。

第一个goroutine是counter,产生一个0,1,2的整数序列，然后通过一个管道发送给第二个goroutine,计算数值的平方，然后将结果通过另一个通道发送给第三个goroutine,接受值并输出它们。

package main

import "fmt"

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

	//counter

	go func() {
		for x := 0; x < 100; x++ {
			naturals <- x
		}
		close(naturals)
	}()

	//squarer
	go func() {
		for x := range naturals {
			squares <- x * x
		}
		close(squares)
	}()
	//printer
	for x := range squares {
		fmt.Println(x)
	}
}

关闭通道还可以作为一个广播机制

  • 有缓冲和无缓冲区别

无缓冲 要求发送和接收必须同时准备好，才可以完成，属于同步操作，不然会发生阻塞等待

有缓冲 可以接受 n 个值，属于异步操作

  • 双向带缓冲区
/*
chan定义方式
变量名 chan = make(chan 数据类型,大小）
ch := make(chan int,10)  双向带缓冲区
fmt.Println(unsafe.Sizeof(ch))  结果为8
*/

package main

import (
	"fmt"
	"unsafe"
)

func main() {
	ch := make(chan int, 10) //双向带缓冲区
	fmt.Println(unsafe.Sizeof(ch))
}
  • 单向 chan
/*
单向chan
ch := make(chan<-  int) 接收
ch := make(<-chan  int) 发送
 */

package main

import (
        "fmt"
        "time"
)

func main(){
        ch :=make(chan int)
        go func(){
                value := <-ch
                fmt.Println(value)
        }()
        ch <- 1
        time.Sleep(time.Second * 2)
}
C:\Users\vSphere\go\src\day04>go run 11.go
1

4.空结构体在 chan 上的应用

  • 空结构体不占用内存空间，通知作用
// 空结构体不占用内存空间，通知作用
package main

import (
	"fmt"
	"time"
)

func main() {
	ch := make(chan struct{})
	go func() {
		time.Sleep(time.Second)
		fmt.Println("协程执行完成。。。")
		close(ch)
	}()
	fmt.Println("主函数执行")
	<-ch
	fmt.Println("主函数执行完成")
}

5.select 和 channel

  • 超时处理
select {
case str := <- resultChan
	fmt.Println("receive str",str)
case <- time.After(time.Second * 5)
	fmt.Println("timeout!!")
}
  • 退出
select {
case <-quitChan:
	cleanUp()
	return
default:
}
  • 判断 channel 是否阻塞
var ch chan int = make(chan int, 5)
select {
case ch <- data:
	fmt.Println("add success")
default: //channel
}

6.并发的同步模型

  • Mutex

开箱即用，不用初始化，将并行操作修改为串行操作

Mutex 不是重入锁，避免多次锁定

推荐使用 defer 进行解锁

不要解锁未锁定的锁 panic

不要重复解锁 panic

不要在多个函数间直接传递，值 copy 以后，即是不同的互斥锁

结构体方法定义的接受对象如果包含了 Mutex，则使用指针值接受对象

// A Mutex is a mutual exclusion lock.
// The zero value for a Mutex is an unlocked mutex.
//
// A Mutex must not be copied after first use.
type Mutex struct {
	state int32
	sema  uint32
}
// A Locker represents an object that can be locked and unlocked.
type Locker interface {
	Lock()
	Unlock()
}
var mutex sync.Mutex
mutex.Lock()
defer mutex.Unlock()

  • sync.RWMutex 读写锁

    • 读读可以并行，读写不能并行

  • sync.WaitGroup

    • 适合于一对多的 goroutine 的协同等待
    • 三个方法：Add Done Wait
    • 底层计数器机制实现，默认值为 0
package main

import (
	"fmt"
	"sync"
)

func main() {
	var mutex sync.Mutex
	counter := 0
	wg := sync.WaitGroup{}
	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func() {
			mutex.Lock()
			defer mutex.Unlock()
			defer wg.Done()
			counter++
		}()
	}
	wg.Wait()
	fmt.Println(counter)
}
推荐阅读