【原创】go语言学习（二十）并发编程

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Go›Go教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

并发和并行

1、概念
A. 并发：同一时间段内执行多个操作。
B. 并行：同一时刻执行多个操作。

Goroutine初探

1、多线程

A. 线程是由操作系统进行管理，也就是处于内核态。
B. 线程之间进行切换，需要发生用户态到内核态的切换。
C. 当系统中运行大量线程，系统会变的非常慢。
D. 用户态的线程，支持大量线程创建。也叫协程或goroutine。

2、创建goroutine

package main
import (
    "fmt"
)
func hello() {
    fmt.Println("Hello world goroutine")
}
func main() {
    go hello()
    fmt.Println("main function")
}

3、修复代码：主进程存在，goroutine才能执行。

package main
import (
    "fmt“
    “time”
)
func hello() {
    fmt.Println("Hello world goroutine")
}
func main() {
    go hello()
    time.Sleep(1*time.Second)
    fmt.Println("main function")
}

Goroutine实战

1、启动多个goroutine

package main
import (
    "fmt"
    "time"
)
func numbers() {
    for i := 1; i <= 5; i++ {
        time.Sleep(250 * time.Millisecond)
        fmt.Printf("%d ", i)
    }
}
func alphabets() {
    for i := 'a'; i <= 'e'; i++ {
        time.Sleep(400 * time.Millisecond)
        fmt.Printf("%c ", i)
    }
}
func main() {
    go numbers()
    go alphabets()
    time.Sleep(3000 * time.Millisecond)
    fmt.Println("main terminated")
}

2、程序分析

3、多核控制

A. 通过runtime包进行多核设置
B. GOMAXPROCS设置当前程序运行时占用的cpu核数
C. NumCPU获取当前系统的cpu核数

package main

import (
	"fmt"
	"time"
)

func hello(i int) {
	fmt.Println("hello goroutine", i)
}

func main() {
	//runtime.GOMAXPROCS(1)
	//fmt.Println(runtime.NumCPU())

	//单线程
	//hello()
	//fmt.Println("mainthread terminate")

	// go 多线程
	//go hello()
	//fmt.Println("mainthread terminate")
	//time.Sleep(time.Second)

	for i := 0; i < 10; i++ {
		go hello(i)
	}
	time.Sleep(time.Second)
}

Goroutine原理浅析

1、模型抽象

A. 操作系统线程： M
B. 用户态线程（goroutine）: G
C. 上下文对象：P

2、goroutine调度

3、系统调用怎么处理

Channel介绍

1、channel介绍

A. 本质上就是一个队列，是一个容器
B. 因此定义的时候，需要只定容器中元素的类型
C. var 变量名 chan 数据类型

package main
import "fmt"
func main() {
    var a chan int
    if a == nil {
        fmt.Println("channel a is nil, going to define it")
        a = make(chan int)
        fmt.Printf("Type of a is %T", a)
    }
}

2、元素入队和出队

A. 入队操作，a <- 100
B. 出队操作：data := <- a

package channel

import "fmt"

// 管道

func main() {
	var c chan int
	fmt.Printf("c=%v", c)

	// 初始化通道int型，10个元素
	c = make(chan int, 10)
	fmt.Printf("c=%v", c)

	// 插入数据
	c <- 100
	/*
		// 读取数据
		data := <- c
		fmt.Printf("data:%v\n", data)
	*/

	// 丢弃元素
	<-c
}

3、阻塞chan

package main
import "fmt"
func main() {
    var a chan int
    if a == nil {
        fmt.Println("channel a is nil, going to define it")
        a = make(chan int)
        a <- 10
        fmt.Printf("Type of a is %T", a)
    }
}

4、使用chan来进行goroutine同步

package main
import (
"fmt"
)
func hello(done chan bool) {
    fmt.Println("Hello world goroutine")
    done <- true
}
func main() {
    done := make(chan bool)
    go hello(done)
    <-done
    fmt.Println("main function")
}

5、使用chan来进行goroutine同步

package main
import (
    "fmt"
    "time"
)
func hello(done chan bool) {
    fmt.Println("hello go routine is going to sleep")
    time.Sleep(4 * time.Second)
    fmt.Println("hello go routine awake and going to write to done")
    done <- true
}
func main() {
    done := make(chan bool)
    fmt.Println("Main going to call hello go goroutine")
    go hello(done)
    <-done
    fmt.Println("Main received data")
}

6、单向chan

package main
import "fmt"
func sendData(sendch chan<- int) {
    sendch <- 10
}
func readData(sendch <-chan int) {
    sendch <- 10
}
func main() {
    chnl := make(chan int)
    go sendData(chnl)
    readData(chn1)
}

7、chan关闭

package main
import (
    "fmt"
)
func producer(chnl chan int) {
    for i := 0; i < 10; i++ {
        chnl <- i
    }
    close(chnl)
}
func main() {
    ch := make(chan int)
    go producer(ch)
    for {
        v, ok := <-ch
        if ok == false {
            break
        }
        fmt.Println("Received ", v, ok)
    }
}

8、 for range操作

package main
import (
    "fmt"
)
func producer(chnl chan int) {
    for i := 0; i < 10; i++ {
        chnl <- i
    }
    close(chnl)
}
func main() {
    ch := make(chan int)
    go producer(ch)
    for v := range ch {
        fmt.Println("Received ",v)
    }
}

9、带缓冲区的chanel

A. Ch := make(chan type, capacity)

package main
import (
    "fmt"
)
func main() {
    ch := make(chan string, 2)
    ch <- “hello"
    ch <- “world"
    fmt.Println(<- ch)
    fmt.Println(<- ch)
}

package main
import (
    "fmt"
    "time"
)
func write(ch chan int) {
    for i := 0; i < 5; i++ {
        ch <- i
        fmt.Println("successfully wrote", i, "to ch")
    }
    close(ch)
}
func main() {
    ch := make(chan int, 2)
    go write(ch)
    time.Sleep(2 * time.Second)
    for v := range ch {
        fmt.Println("read value", v,"from ch")
        time.Sleep(2 * time.Second)
    }
}

10、channel的长度和容量

A. Ch := make(chan type, capacity)

package main
import (
    "fmt"
)
func main() {
    ch := make(chan string, 3)
    ch <- "naveen"
    ch <- "paul"
    fmt.Println("capacity is", cap(ch))
    fmt.Println("length is", len(ch))
    fmt.Println("read value", <-ch)
    fmt.Println("new length is", len(ch))
}

Waitgroup介绍

1、如何等待一组goroutine结束？

A. 方法一，使用不带缓冲区的channel实现

package main
import (
    "fmt"
    "time"
)
func process(i int, ch chan bool) {
    fmt.Println("started Goroutine ", i)
    time.Sleep(2 * time.Second)
    fmt.Printf("Goroutine %d ended\n", i)
    ch <- true
}
func main() {
    no := 3
    exitChan := make(chan bool, no)
    for i := 0; i < no; i++ {
        go process(i, exitChan)
    }
    for i := 0; I < no;i++{
        <-exitChan
    }
    fmt.Println("All go routines finished executing")
}

B. 方法二，使用sync. WaitGroup实现

package main
import (
    "fmt"
    "sync"
    "time"
)
func process(i int, wg *sync.WaitGroup) {
    fmt.Println("started Goroutine ", i)
    time.Sleep(2 * time.Second)
    fmt.Printf("Goroutine %d ended\n", i)
        wg.Done()
    }
func main() {
    no := 3
    var wg sync.WaitGroup
    for i := 0; i < no; i++ {
        wg.Add(1)
        go process(i, &wg)
    }
    wg.Wait()
    fmt.Println("All go routines finished executing")
}

Workerpool的实现

1、worker池的实现

A. 生产者、消费者模型，简单有效
B. 控制goroutine的数量，防止goroutine泄露和暴涨
C. 基于goroutine和chan，构建workerpool非常简单

package mail

import (
	"fmt"
	"math/rand"
)

// worker生产者消费者模型
type Job struct {
	Number int
	Id int
}

type Result struct {
	job *Job
	sum int
}

func calc(job *Job, result chan *Result){
	var sum int
	number := job.Number
	for number != 0 {
		tmp := number % 10
		sum += tmp
		number /= 10
	}

	r := &Result{
		job: job,
		sum: sum,

	}

	result <- r
}

func Worker(){
	for job:= range jobChan {
		calc(job, resultChan)
	}
}

func startWorkerPool(num int, JobChan chan *Job, resultChan *Result){
	for i := 0; i < num; i++ {
		go Worker(JobChan, resultChan)
	}
}

func printResult(resultChan chan*Result) {
	for result := range resultChan {
		fmt.Printf("job id:%v number:%v result:%d\n",result.job.Id, result.job.Number, result.sum)
	}
}

func main(){
	jobChan := make(chan *Job, 1000)
	resultChan := make(chan *Result, 1000)

	startWorkerPool(128, jobChan, resultChan)

	for i := 0; i < 128; i ++ {
		go calc()
	}

	go printResult(resultChan)
	var id int
	for {
		id++
		number := rand.Int()
		job := &Job {
			Id: id,
			Number: number,
		}

		jobChan <- job
	}
}

2、项目需求分析

A. 计算一个数字的各个位数之和，比如123，和等于1+2+3=6
B. 需要计算的数字使用随机算法生成

3、方案介绍

A. 任务抽象成一个个job
B. 使用job队列和result队列
C. 开一组goroutine进行实际任务计算，并把结果放回result队列