goroutine和channel
goroutine
多线程
func hello() {
//fmt.Printf("Hello Goroutine!!\n")
for i:=0;i<100;i++ {
fmt.Printf("hello:%d\n",i)
time.Sleep(time.Millisecond)
}
}
func main() {
go hello() //启动了一个独立的线程,使其与下面的代码交替执行，使之成为一个多线程
//fmt.Printf("main function\n")
for i:=0;i<100;i++ {
fmt.Printf("main:%d\n",i)
time.Sleep(time.Millisecond)
}
time.Sleep(time.Second) //修复代码，使得主线程退出的时候子线程能执行
}多个goroutine
func nunmers() {
for i :=0;i<=5;i++ {
time.Sleep(time.Millisecond*250)
fmt.Printf("%d\n",i)
}
}
func chars() {
for i:=‘a‘;i<=‘e‘;i++ {
time.Sleep(time.Millisecond*400)
fmt.Printf("%c\n",i)
}
}
func main() {
go nunmers()
go chars()
time.Sleep(time.Second*3)
}进程和线程
- 进程：
- 进程是程序在操作系统中的一次执行过程，系统进行资源分配和调度的一个独立单位
- 线程：
- 线程是进程的一个执行实体，是CPU调度和分派的基本单位，他是比进程更小的能独立运行的基本单位
- 一个进程可以创建和撤销多个线程，同一个进程中的多个线程之间可以并发执行
并发与并行
- 多线程程序在一个核的CPU上运行，这是并发
- 多线程程序在多个核的CPU上运行，这是并行
协程和线程
- 协程： 独立的栈空间，共享堆空间，调度由用户自己控制，本质上有点类似于用户级线程，这些用户级线程的调度也是自己实现的
- 线程： 一个县城上可以跑多个协程，协程是轻量级的线程
goroutine的调度模型
- M（线程）P（上下文）G（goroutine)
设置golang运行的CPU核心数
func main() {
cpu := runtime.NumCPU()
//限制核心数（新版本不用考虑），比如监控程序，可以控制其运行的资源消耗
//runtime.GOMAXPROCS(1)
for i:=0;i <=8;i++ {
go func() {
}()
}
fmt.Printf("%d\n",cpu)
time.Sleep(time.Second*12)
}channel
- 不同的goroutine之间如何进行通讯：
- 全局变量和锁同步： 不推荐
- Channel： 先进先出的队列
- channel的概念
- 类似于unix中的管道
- 先进先出
- 线程安全，多个goroutine同事访问不需要加锁
- channel是有类型的，一个证书的channel只能整数
channel的声明
- 引用类型，需要使用make来初始化
var 变量名 chan 类型
var test chan int
var test chan string
var test chan map[string]string
var test chan stu
var test chan *stu- channel的初始化与读写
func main() {
//var intChan chan int = make(chan int,1) 有缓冲区的channel
var intChan chan int = make(chan int) // 无缓冲区channel
fmt.Printf("intChanin:%p\n",intChan)
go func() {
//写入数据
intChan <- 100
fmt.Printf("insert item end\n")
}()
go func() {
//读取数据
fmt.Printf("start\n")
time.Sleep(time.Second*3)
var a int
a = <- intChan
fmt.Printf("intChan:%d\n",a)
}()
time.Sleep(time.Second*5)
}goroutine和channel相结合
- 生产者消费者模型
func senData(ch chan string) {
ch <- "Washinton"
ch <- "Tripoli"
ch <- "LongDong"
ch <- "Beijing"
ch <- "Tokyo"
}
func getData(ch chan string) {
var input string
for {
input = <- ch
fmt.Printf("input=%s\n",input)
}
}
func main() {
ch := make(chan string,10)
go senData(ch)
go getData(ch)
time.Sleep(time.Second*1)
}阻塞channel的情况
- 无写入，空读取会阻塞(空channel)
- 写入的数量超过缓冲区也会阻塞 （channel满了）func sendChans(ch chan string) {
var i int
for {
var str string
str = fmt.Sprintf("stu %d\n",i)
fmt.Printf("write:%s\n",str)
//无人消费数据，则阻塞,如果无缓冲区的话，写入也是阻塞的
ch <- str
i++
}
}
func main() {
ch := make(chan string,10) //带缓冲区
//ch := make(chan string) //无缓冲区
go sendChans(ch)
time.Sleep(time.Second*200)
}channel之间的同步
- 新增退出检测的方式保证同步
func sendChannels(ch chan string,exitCh chan bool) {
ch <- "AA"
ch <- "BB"
ch <- "CC"
ch <- "DD"
ch <- "EE"
close(ch)
exitCh <- true
}
func getChannels(ch chan string,exitCh chan bool) {
for {
// 检查channel关闭
input,ok := <- ch
if !ok {
break
}
fmt.Printf("getchannels中的input:%s\n",input)
}
exitCh <- true
}
func getChannels2(ch chan string,exitCh chan bool) {
for {
// 检查channel关闭
input,ok := <- ch
if !ok {
break
}
fmt.Printf("getchannels中的input:%s\n",input)
}
exitCh <- true
}
func main() {
ch := make(chan string)
exitChan := make(chan bool,3)
go sendChannels(ch,exitChan)
go getChannels(ch,exitChan)
go getChannels2(ch,exitChan)
//三次检查状态，然后退出,不需要time.sleep，等待其他goroutine退出
<- exitChan //取出元素，然后扔掉
<- exitChan
<- exitChan
}- 使用waitGroup的方式确保goroutine的同步
func SendChannels1(ch chan string,waitGroup *sync.WaitGroup) {
ch <- "AA"
ch <- "BB"
ch <- "CC"
ch <- "DD"
ch <- "EE"
close(ch)
fmt.Printf("send data exited\n")
waitGroup.Done() //结束goroutine的标志，然后对标志位的数字-1
}
func GetChannels1(ch chan string,waitGroup *sync.WaitGroup) {
for {
input,ok := <- ch
if !ok{
break
}
fmt.Printf("getchannels中的input:%s\n",input)
}
fmt.Printf("get data exited\n")
waitGroup.Done()
}
func GetChannels2(ch chan string,waitGroup *sync.WaitGroup) {
for {
input,ok := <- ch
if !ok{
break
}
fmt.Printf("getchannels中的input:%s\n",input)
}
fmt.Printf("get data2 exited\n")
waitGroup.Done()
}
func main() {
var wg sync.WaitGroup
ch := make(chan string)
wg.Add(3) //每执行完一次goroutine就会-1
go SendChannels1(ch,&wg)
go GetChannels1(ch,&wg)
go GetChannels2(ch,&wg)
wg.Wait() //等到运行完成之后返回
fmt.Printf("main goroutine exited\n")
}- 关闭channel的时候，生产者channel的数据是保存的，不丢失
func main() {
intChan := make(chan int,10)
for i:=0;i<10;i++ {
intChan<- i
}
//关闭channel
close(intChan)
time.Sleep(time.Second)
for j:=0;j<10;j++ {
a := <- intChan
fmt.Printf("a=%d\n",a)
}
}- channel的遍历
func GetChannels2(ch chan string,waitGroup *sync.WaitGroup) {
//for range会自动判断channel是否关闭
for v :=range ch {
fmt.Printf("get data2 %s\n",v)
}
fmt.Printf("get data2 exited\n")
waitGroup.Done()
}- chan的关闭
- 使用内置函数close进行关闭，chan关闭之后，for range 遍历chan中已经存在的元素后结束
- 使用内置函数close进行关闭，chan关闭之后，没有使用for range的写法，需要使用
v,ok :=<-chan判断chan是否关闭
chan的只读和只写
需要注意<-是在chan关键字的位置，<-在chan左侧表示只读，在右侧表示只写
func chanPerms() {
var readOnly <- chan int = make(chan int,100)
// readOnly <- 100 只读不可写
var writeOnly chan <- int = make(chan int,10)
// <- writeOnly 只写不可读
}- 使用场景： 三方调用只读只写权限控制，防止误操作
对chan进行select操作
func main() {
var intChan chan int = make(chan int,10)
var strChan chan string = make(chan string,10)
var wg sync.WaitGroup
wg.Add(2)
// 插入数据
go func() {
var count int
for count < 1000 {
count++
select {
case intChan <- 10:
fmt.Printf("write to int chan succ\n")
case strChan <- "hello":
fmt.Printf("write to str chan succ\n")
default:
fmt.Printf("all chan is full\n")
time.Sleep(time.Second)
}
}
wg.Done()
}()
wg.Wait()
//读取数据
go func() {
var count int
for count < 10000 {
count ++
select {
case a := <- intChan:
fmt.Printf("read from int chain a:%d\n",a)
case <- strChan:
fmt.Printf("read from str chan\n")
default:
fmt.Printf("all chan is empty\n")
time.Sleep(time.Second)
}
}
wg.Done()
}()
wg.Wait()
}原文地址：https://blog.51cto.com/13812615/2483168
时间： 2024-07-31 13:01:46