go 并发 gorouting chan channel select Mutex sync.One

news2025/2/22 12:34:46

goroutine

// head： 前缀  index：是一个int的指针
func print(head string, index *int) {
    for i := 0; i < 5; i++ {
       // 指针对应的int ++
       *index++
       fmt.Println(*index, head, i)
       // 暂停1s
       time.Sleep(1 * time.Second)
    }
}

/*
Go 允许使用 go 语句开启一个新的运行期线程，即 goroutine
以一个不同的、新创建的 goroutine 来执行一个函数
同一个程序中的所有 goroutine 共享同一个地址空间。
*/
func main() {
    fmt.Println("main ...")
    index := 0
    go print("first", &index)
    go print("second", &index)
    time.Sleep(6 * time.Second)
    fmt.Println("success ...")
}

chan 一般用法

// 求和，并将数据放在channel中
func sum(arr []int, resultChan chan int) {
    if len(arr) <= 0 {
       resultChan <- 0
       return
    }

    var sum = 0
    for _, value := range arr {
       sum += value
    }

    fmt.Println(sum)

    // 将结果放在channel中
    resultChan <- sum
}

/*
channel 用于 goroutine之间进行通信
 1. 创建channel
    ch1 := make(chan 类型)  默认是没有缓冲区的
    ch2 := make(chan 类型, 缓存长度)
 2. 添加数据到channel中
    ch1 <- 123
 3. 从channel中获取数据
    var value = <- ch1

有无缓冲区区别：
1. 没有缓冲区 按照缓冲区为1来处理，即channel只能放一个数据
2. channel满了后就会阻塞，直到有空位才可以继续放入数据
3. 获取数据类似，阻塞到channel中有数据
*/
func main() {
    fmt.Println("main ...")
    // 创建一个没有缓冲区的channel
    resultChan := make(chan int)
    array1 := []int{10, 20, 30}
    array2 := []int{1, 2, 3}
    // 给两个数组求和并将结果放在channel中
    go sum(array1, resultChan)
    go sum(array2, resultChan)
    // 从channel中获取两个数据，打印到console
    fmt.Println(<-resultChan, <-resultChan)
    fmt.Println("continue ...")
    // 如果继续获取则会报错：fatal error: all goroutines are asleep - deadlock!
    // fmt.Println(<-resultChan)
    fmt.Println("success ...")
}

无缓冲区的chan只能结合goroutine使用

func main() {
    fmt.Println("main ...")
    // 创建一个没有缓冲区的channel
    resultChan := make(chan int)
    // chan 只能结合goroutine来使用，否则报错
    // fatal error: all goroutines are asleep - deadlock!
    resultChan <- 10
    fmt.Println(<-resultChan)
    fmt.Println("success ...")
}

有缓冲区的chan可直接赋值

func main() {
    fmt.Println("main ...")
    // 创建一个有缓冲区的channel
    ch := make(chan int, 2)
    ch <- 1
    ch <- 2
    fmt.Println(<-ch)
    fmt.Println(<-ch)
    fmt.Println("success ...")
}

for rang获取channel数据

func addData(ch chan int, len int) {
    for i := 0; i < len; i++ {
       ch <- i
    }
    // 如果不关闭，for val := range ch 会阻塞获取数据
    close(ch)
}

/*
  - 可以使用rang遍历channel，如果channel关闭则直接结束，否则会阻塞等待数据的输入
    for val := range ch
*/
func main() {
    fmt.Println("main ...")
    // 创建一个有缓冲区的channel
    len := 5
    ch := make(chan int, len)
    go addData(ch, len)
    for val := range ch {
       fmt.Println(val)
    }
    fmt.Println("success ...")
}

select **等待多个goroutine

select可以等待多个goroutine，会阻塞一直到某个case不在阻塞。

func print1(header string, ch1, ch2 chan int) {
    for i := 0; i < len; i++ {
       select {
          case val := <-ch1:
             fmt.Println(header, val)
             time.Sleep(time.Second)
          case ch2 <- i:
             // do nothing
       }
    }
}

func main() {
    fmt.Println("main ...")
    ch1 := make(chan int)
    ch2 := make(chan int)
    go print1("f1", ch1, ch2)
    go print1("f2", ch2, ch1)
    time.Sleep(6 * time.Second)
    fmt.Println("success ...")
}

WaitGroup等待所有goroutine完成

类似java中的CountDownLatch

// 不怎么理解为什么group要用指针
func work(index int, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println(index)
    time.Sleep(time.Second)
}

func main() {
    fmt.Println("main ...")

    var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
       wg.Add(1)
       // 为什么要将wg的指针传过去
       go work(i, &wg)
    }

    // 阻塞到所有的goroutine完成后
    wg.Wait()

    fmt.Println("success ...")
}

并发锁Mutex

type ConcurrentMap struct {
    lock    sync.Mutex
    hashmap map[string]int
}

// 1. cm *ConcurrentMap 要传指针，否则操作的是副本
// 2. wg *sync.WaitGroup 这个也传指针，确保操作的是一个对象
func (cm *ConcurrentMap) inc(index int, key string, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println(index, "start")
    cm.lock.Lock()
    before := cm.hashmap[key]
    fmt.Println(index, "before", before)
    cm.hashmap[key] = cm.hashmap[key] + 1
    time.Sleep(time.Microsecond * 100)
    after := cm.hashmap[key]
    fmt.Println(index, "after", after)
    if before+1 != after {
       fmt.Println(index, "error >>>>>")
    }
    cm.lock.Unlock()
    fmt.Println(index, "end")
}

func main() {
    fmt.Println("main ...")

    cm := ConcurrentMap{hashmap: make(map[string]int)}
    var wg sync.WaitGroup
    for i := 0; i < 3; i++ {
       wg.Add(1)
       go cm.inc(i, "apple", &wg)
    }
    wg.Wait()
    fmt.Println("success ...", cm.hashmap["apple"])
}

RWMutex 读写锁子

参考：

https://www.jianshu.com/p/679041bdaa39

sync.Once 配置文件只加载一次

需求：获取配置文件，如果没有价值只就加载

写法1：

func initMap() {
    if hasInitMap {
       return
    }
    initMapLock.Lock()
    defer initMapLock.Unlock()
    if !hasInitMap {
       fmt.Println("init map")
       m = map[string]string{
          "aaa": "111",
          "bbb": "22",
       }
       hasInitMap = true
    }
}

func getValue1(key string) string {
    initMap()
    return m[key]
}

写法2：

// 定义一次执行对象
var once sync.Once

func initMap2() {
    m = map[string]string{
       "aaa": "111",
       "bbb": "22",
    }
}

func getValue2(key string) string {
    // 一次执行
    once.Do(initMap2)
    return m[key]
}

func main() {
    fmt.Println("main ...")
    for i := 0; i < 20; i++ {
       fmt.Println(getValue1("aaa"))
       fmt.Println(getValue2("aaa"))
    }
    fmt.Println("success ...")
}