goroutine之间的同步

goroutine是golang中在语言级别实现的轻量级线程,仅仅利用go就能立刻起一个新线程。多线程会引入线程之间的同步问题,经典的同步问题如生产者-消费者问题,在c,java级别需要使用锁、信号量进行共享资源的互斥使用和多线程之间的时序控制,而在golang中有channel作为同步的工具。

1. channel实现两个goroutine之间的通信

package main

import "strconv"
import "fmt" func main() {
taskChan := make(chan string, 3)
doneChan := make(chan int, 1) for i := 0; i < 3; i++ {
taskChan <- strconv.Itoa(i)
fmt.Println("send: ", i)
} go func() {
for i := 0; i < 3; i++ {
task := <-taskChan
fmt.Println("received: ", task)
}
doneChan <- 1
}() <-doneChan
}
make(chan TYPE {, NUM})CHAN <- DATAgo func(){ BODY }()DATA := <-CHAN

上面这个程序就是main routine向另一个routine发送了3条int类型的数据,当3条数据被接收到后,主线程也从阻塞状态恢复运行,随后结束。

2. 不要陷入“死锁”

我一开始用channel的时候有报过"fatal error: all goroutines are asleep - deadlock! "的错误,真实的代码是下面这样的:

package main
import "fmt" func main() {
ch := make(chan int)
ch <- 1 // I'm blocked because there is no channel read yet.
fmt.Println("send")
go func() {
<-ch // I will never be called for the main routine is blocked!
fmt.Println("received")
}()
fmt.Println("over")
}

我的本意是从main routine发送给另一个routine一个int型的数据,但是运行出了上述的错误,原因有2个:

ch <- 1

解决这个“死锁”的方法可是是设置channel的容量大小大于1,那么channel就不会因为数据输入而阻塞主程; 或者将数据输入channel的语句置于启动新的goroutine之后。

3. channel作为状态转移的信号源

我跟着MIT的分布式计算课程做了原型为Map-Reduce的课后练习,目的是实现一个Master向Worker分派任务的功能:Master服务器去等待Worker服务器连接注册,Master先将Map任务和Reduce任务分派给这些注册Worker,等待Map任务全部完成,然后将Reduce任务再分配出去,等待全部完成。

// Initialize a channel which records the process of the map jobs.
mapJobChannel := make(chan string) // Start a goroutine to send the nMap(the number of the Map jobs) tasks to the main routine.
go func() {
for m := 0; m < nMap; m++ {
// Send the "start a Map job <m>" to the receiver.
mapJobChannel <- "start, " + strconv.Itoa(m)
}
}() // Main routine listens on this mapJobChannel for the Map job task information.
nMapJobs := nMap // Process the nMap Map tasks util they're all done.
for nMapJobs > 0 {
// Receive the Map tasks from the channel.
taskInfo := strings.Split(<-mapJobChannel, ",")
state, mapNo := taskInfo[0], taskInfo[1] if state == "start" {
// Assign a Map task with number mapNo to a worker.
go func() {
// Wait for a worker to finish the Map task.
ok, workNo := assignJobToWorker("Map", mapNo)
if ok {
// Send a task finish signal and set the worker's state to idle.
mapJobChannel <- "end, " + mapNo
setWorkerState(workNo, "idle")
} else {
// Restart this task and set the worker's state to finished.
mapJobChannel <- "start, " + mapNo
setWorkerState(workNo, "finished")
}
}()
} else {
nMapJobs--
}
}

以上是我截取自己写的代码,关于用channel来传递当前Map任务的进度信息,用类似信号的方式标注当前的任务执行状态。

  • 当从channel中读取到"start, {NUM}"时找一个空闲的Worker去执行Map任务,并且等待它的完成,完成成功则向channel中发送"end, {NUM}"信号,表明任务完成,如果失败,就重发"start, {NUM}"信号。
  • 从channel中读取到"end, {NUM}"时,把剩余任务数减1。

    这种信号触发的方式,触发Master的状态转移,并且可以通过增加信号以及信号处理的方式,拓展业务处理的情况,暂时还能处理这个需求情景。